diff --git a/internal/strategy/git/export_test.go b/internal/strategy/git/export_test.go index 528bee2..d859ad5 100644 --- a/internal/strategy/git/export_test.go +++ b/internal/strategy/git/export_test.go @@ -71,3 +71,11 @@ func ColdPreparationQueueLimitForTest() int { func (s *Strategy) ClaimSnapshotForTest(upstream string) (bool, error) { return s.snapshotCoord.Claim(snapshotJobBase, upstream, 0) } + +// HoldUploadPackCloneSlotForTest acquires a clone-shaped upload-pack slot +// and returns a release func. admitted is false if the limiter rejected or +// the context was cancelled. +func (s *Strategy) HoldUploadPackCloneSlotForTest(ctx context.Context, repo string) (release func(), admitted bool) { + release, result, _ := s.uploadPackLimiter.acquire(ctx, repo) + return release, result == acquireOK +} diff --git a/internal/strategy/git/git.go b/internal/strategy/git/git.go index 75be8a3..36ef4ee 100644 --- a/internal/strategy/git/git.go +++ b/internal/strategy/git/git.go @@ -53,6 +53,14 @@ type Config struct { BundleCacheTTL time.Duration `hcl:"bundle-cache-ttl,optional" help:"TTL of cached server-side git bundles." default:"2h"` SnapshotFilters map[string]string `hcl:"snapshot-filters,optional" help:"Per-repository git partial-clone filter applied to workstation snapshots, keyed by host/org/repo (e.g. {\"github.com/org/repo\": \"blob:none\"}). Filtered snapshots contain full history metadata but only the blobs needed for the HEAD checkout; clients lazily fetch historical blobs through cachew on demand."` + + // Clone-shaped git-upload-pack gating. Incremental fetches (pkt-lines + // containing "have ") are never limited. 0 concurrency is unlimited and + // matches historical behaviour. + UploadPackCloneConcurrency int `hcl:"upload-pack-clone-concurrency,optional" help:"Max concurrent clone-shaped git-upload-pack responses (no 'have' lines). 0 is unlimited." default:"0"` + UploadPackClonePerRepoConcurrency int `hcl:"upload-pack-clone-per-repo-concurrency,optional" help:"Max concurrent clone-shaped git-upload-pack responses per repository. 0 is unlimited." default:"0"` + UploadPackCloneQueueTimeout time.Duration `hcl:"upload-pack-clone-queue-timeout,optional" help:"How long a clone-shaped upload-pack may wait for a slot before 503. 0 fails immediately when at capacity." default:"0"` + UploadPackCloneRetryAfter time.Duration `hcl:"upload-pack-clone-retry-after,optional" help:"Retry-After value sent with 503 overload responses." default:"30s"` } type Strategy struct { @@ -75,6 +83,7 @@ type Strategy struct { mirrorPreparations sync.Map // One entry per upstream URL prevents duplicate preparation jobs. deferredRestoreOnce sync.Map // keyed by upstream URL, ensures at most one deferred restore per repo metrics *gitMetrics + uploadPackLimiter *uploadPackLimiter repoCounts *RepoCounts snapshotCoord *SnapshotCoordinator coldPreparationDelay func() time.Duration @@ -141,16 +150,17 @@ func New( m := newGitMetrics() s := &Strategy{ - config: config, - cache: cache, - cloneManager: cloneManager, - httpClient: http.DefaultClient, - ctx: ctx, - scheduler: scheduler.WithQueuePrefix("git"), - spools: make(map[string]*RepoSpools), - tokenManager: tokenManager, - metrics: m, - metadataWired: make(chan struct{}), + config: config, + cache: cache, + cloneManager: cloneManager, + httpClient: http.DefaultClient, + ctx: ctx, + scheduler: scheduler.WithQueuePrefix("git"), + spools: make(map[string]*RepoSpools), + tokenManager: tokenManager, + metrics: m, + uploadPackLimiter: newUploadPackLimiter(config), + metadataWired: make(chan struct{}), coldPreparationDelay: func() time.Duration { return rand.N(coldPreparationSpread) //nolint:gosec // The delay does not protect sensitive data. }, @@ -221,7 +231,11 @@ func New( mux.Handle("GET /git/{host}/{path...}", http.HandlerFunc(s.handleRequest)) mux.Handle("POST /git/{host}/{path...}", http.HandlerFunc(s.handleRequest)) - logger.InfoContext(ctx, "Git strategy initialized", "snapshot_interval", config.SnapshotInterval) + logger.InfoContext(ctx, "Git strategy initialized", + "snapshot_interval", config.SnapshotInterval, + "upload_pack_clone_concurrency", config.UploadPackCloneConcurrency, + "upload_pack_clone_per_repo_concurrency", config.UploadPackClonePerRepoConcurrency, + "upload_pack_clone_queue_timeout", config.UploadPackCloneQueueTimeout) return s, nil } @@ -387,6 +401,11 @@ func (s *Strategy) handleGitRequest(w http.ResponseWriter, r *http.Request, host if err := s.repoCounts.IncrementClone(upstreamURL); err != nil { logger.WarnContext(ctx, "Failed to increment repo clone count", "error", err) } + release, admitted := s.gateCloneUploadPack(w, r, upstreamURL) + if !admitted { + return + } + defer release() } state := repo.State() diff --git a/internal/strategy/git/metrics.go b/internal/strategy/git/metrics.go index f19ede3..c11d686 100644 --- a/internal/strategy/git/metrics.go +++ b/internal/strategy/git/metrics.go @@ -13,47 +13,51 @@ import ( ) type gitMetrics struct { - operationDuration metric.Float64Histogram - operationTotal metric.Int64Counter - requestTotal metric.Int64Counter - snapshotServeTotal metric.Int64Counter - snapshotServeSize metric.Float64Histogram - snapshotServeDuration metric.Float64Histogram - bundleServeTotal metric.Int64Counter - bundleServeSize metric.Float64Histogram - bundleServeDuration metric.Float64Histogram - ensureRefsTotal metric.Int64Counter - ensureRefsDuration metric.Float64Histogram - spoolWriterDuration metric.Float64Histogram - spoolFollowerWaitTotal metric.Int64Counter - spoolFollowerWait metric.Float64Histogram - repackPackCount metric.Float64Histogram - snapshotServeBandwidth metric.Float64Histogram - lfsPhaseDuration metric.Float64Histogram - lfsPhaseBytes metric.Float64Histogram + operationDuration metric.Float64Histogram + operationTotal metric.Int64Counter + requestTotal metric.Int64Counter + snapshotServeTotal metric.Int64Counter + snapshotServeSize metric.Float64Histogram + snapshotServeDuration metric.Float64Histogram + bundleServeTotal metric.Int64Counter + bundleServeSize metric.Float64Histogram + bundleServeDuration metric.Float64Histogram + ensureRefsTotal metric.Int64Counter + ensureRefsDuration metric.Float64Histogram + spoolWriterDuration metric.Float64Histogram + spoolFollowerWaitTotal metric.Int64Counter + spoolFollowerWait metric.Float64Histogram + repackPackCount metric.Float64Histogram + snapshotServeBandwidth metric.Float64Histogram + lfsPhaseDuration metric.Float64Histogram + lfsPhaseBytes metric.Float64Histogram + uploadPackCloneGateTotal metric.Int64Counter + uploadPackCloneQueueWait metric.Float64Histogram } func newGitMetrics() *gitMetrics { meter := otel.Meter("cachew.git") return &gitMetrics{ - operationDuration: metrics.NewHistogram(meter, "cachew.git.operation_duration_seconds", "s", "Duration of git operations (clone, fetch, repack, snapshot)", metrics.LatencyBuckets()), - operationTotal: metrics.NewMetric[metric.Int64Counter](meter, "cachew.git.operations_total", "{operations}", "Total number of git operations"), - requestTotal: metrics.NewMetric[metric.Int64Counter](meter, "cachew.git.requests_total", "{requests}", "Total number of git HTTP requests by type"), - snapshotServeTotal: metrics.NewMetric[metric.Int64Counter](meter, "cachew.git.snapshot_serves_total", "{serves}", "Snapshot serve events by source (cache, spool, cold_cache, generated) and repository"), - snapshotServeSize: metrics.NewHistogram(meter, "cachew.git.snapshot_serve_bytes", "By", "Size of served snapshots in bytes", metrics.ByteBuckets()), - snapshotServeDuration: metrics.NewHistogram(meter, "cachew.git.snapshot_serve_duration_seconds", "s", "Wall-clock duration of snapshot serves, from handler entry to last byte sent", metrics.LatencyBuckets()), - bundleServeTotal: metrics.NewMetric[metric.Int64Counter](meter, "cachew.git.bundle_serves_total", "{serves}", "Bundle serve events by source (cache, generated, up_to_date, miss_bad_base, miss) and repository"), - bundleServeSize: metrics.NewHistogram(meter, "cachew.git.bundle_serve_bytes", "By", "Size of served bundles in bytes", metrics.ByteBuckets()), - bundleServeDuration: metrics.NewHistogram(meter, "cachew.git.bundle_serve_duration_seconds", "s", "Wall-clock duration of bundle serves, including any on-demand generation", metrics.LatencyBuckets()), - ensureRefsTotal: metrics.NewMetric[metric.Int64Counter](meter, "cachew.git.ensure_refs_total", "{requests}", "EnsureRefs requests by fetched and status"), - ensureRefsDuration: metrics.NewHistogram(meter, "cachew.git.ensure_refs_duration_seconds", "s", "Duration of EnsureRefs requests, including any upstream fetch", metrics.FastLatencyBuckets()), - spoolWriterDuration: metrics.NewHistogram(meter, "cachew.git.spool_writer_duration_seconds", "s", "Time the snapshot spool writer spent producing the stream", metrics.LatencyBuckets()), - spoolFollowerWaitTotal: metrics.NewMetric[metric.Int64Counter](meter, "cachew.git.spool_follower_waits_total", "{waits}", "Snapshot spool follower events, by outcome (served, writer_failed)"), - spoolFollowerWait: metrics.NewHistogram(meter, "cachew.git.spool_follower_wait_seconds", "s", "Time a snapshot spool follower spent waiting for the writer to publish headers", metrics.FastLatencyBuckets()), - repackPackCount: metrics.NewHistogram(meter, "cachew.git.repack_pack_count", "{packs}", "Pack file count observed before and after repack, by stage (before, after)", metrics.SmallCountBuckets()), - snapshotServeBandwidth: metrics.NewHistogram(meter, "cachew.git.snapshot_serve_bandwidth_mbps", "MiBy/s", "Per-request snapshot serve throughput in MiB/s, by source and repository", metrics.BandwidthMbpsBuckets()), - lfsPhaseDuration: metrics.NewHistogram(meter, "cachew.git.lfs_phase_duration_seconds", "s", "Duration of an LFS-snapshot generation phase (discover, clone, fetch, archive_upload), by status and repository", metrics.LatencyBuckets()), - lfsPhaseBytes: metrics.NewHistogram(meter, "cachew.git.lfs_phase_bytes", "By", "Bytes processed in an LFS-snapshot generation phase, by phase and repository (e.g. .git/lfs size after fetch)", metrics.ByteBuckets()), + operationDuration: metrics.NewHistogram(meter, "cachew.git.operation_duration_seconds", "s", "Duration of git operations (clone, fetch, repack, snapshot)", metrics.LatencyBuckets()), + operationTotal: metrics.NewMetric[metric.Int64Counter](meter, "cachew.git.operations_total", "{operations}", "Total number of git operations"), + requestTotal: metrics.NewMetric[metric.Int64Counter](meter, "cachew.git.requests_total", "{requests}", "Total number of git HTTP requests by type"), + snapshotServeTotal: metrics.NewMetric[metric.Int64Counter](meter, "cachew.git.snapshot_serves_total", "{serves}", "Snapshot serve events by source (cache, spool, cold_cache, generated) and repository"), + snapshotServeSize: metrics.NewHistogram(meter, "cachew.git.snapshot_serve_bytes", "By", "Size of served snapshots in bytes", metrics.ByteBuckets()), + snapshotServeDuration: metrics.NewHistogram(meter, "cachew.git.snapshot_serve_duration_seconds", "s", "Wall-clock duration of snapshot serves, from handler entry to last byte sent", metrics.LatencyBuckets()), + bundleServeTotal: metrics.NewMetric[metric.Int64Counter](meter, "cachew.git.bundle_serves_total", "{serves}", "Bundle serve events by source (cache, generated, up_to_date, miss_bad_base, miss) and repository"), + bundleServeSize: metrics.NewHistogram(meter, "cachew.git.bundle_serve_bytes", "By", "Size of served bundles in bytes", metrics.ByteBuckets()), + bundleServeDuration: metrics.NewHistogram(meter, "cachew.git.bundle_serve_duration_seconds", "s", "Wall-clock duration of bundle serves, including any on-demand generation", metrics.LatencyBuckets()), + ensureRefsTotal: metrics.NewMetric[metric.Int64Counter](meter, "cachew.git.ensure_refs_total", "{requests}", "EnsureRefs requests by fetched and status"), + ensureRefsDuration: metrics.NewHistogram(meter, "cachew.git.ensure_refs_duration_seconds", "s", "Duration of EnsureRefs requests, including any upstream fetch", metrics.FastLatencyBuckets()), + spoolWriterDuration: metrics.NewHistogram(meter, "cachew.git.spool_writer_duration_seconds", "s", "Time the snapshot spool writer spent producing the stream", metrics.LatencyBuckets()), + spoolFollowerWaitTotal: metrics.NewMetric[metric.Int64Counter](meter, "cachew.git.spool_follower_waits_total", "{waits}", "Snapshot spool follower events, by outcome (served, writer_failed)"), + spoolFollowerWait: metrics.NewHistogram(meter, "cachew.git.spool_follower_wait_seconds", "s", "Time a snapshot spool follower spent waiting for the writer to publish headers", metrics.FastLatencyBuckets()), + repackPackCount: metrics.NewHistogram(meter, "cachew.git.repack_pack_count", "{packs}", "Pack file count observed before and after repack, by stage (before, after)", metrics.SmallCountBuckets()), + snapshotServeBandwidth: metrics.NewHistogram(meter, "cachew.git.snapshot_serve_bandwidth_mbps", "MiBy/s", "Per-request snapshot serve throughput in MiB/s, by source and repository", metrics.BandwidthMbpsBuckets()), + lfsPhaseDuration: metrics.NewHistogram(meter, "cachew.git.lfs_phase_duration_seconds", "s", "Duration of an LFS-snapshot generation phase (discover, clone, fetch, archive_upload), by status and repository", metrics.LatencyBuckets()), + lfsPhaseBytes: metrics.NewHistogram(meter, "cachew.git.lfs_phase_bytes", "By", "Bytes processed in an LFS-snapshot generation phase, by phase and repository (e.g. .git/lfs size after fetch)", metrics.ByteBuckets()), + uploadPackCloneGateTotal: metrics.NewMetric[metric.Int64Counter](meter, "cachew.git.upload_pack_clone_gate_total", "{requests}", "Clone-shaped git-upload-pack admissions, rejections, and cancellations"), + uploadPackCloneQueueWait: metrics.NewHistogram(meter, "cachew.git.upload_pack_clone_queue_wait_seconds", "s", "Time spent waiting for a clone-shaped upload-pack slot", metrics.FastLatencyBuckets()), } } @@ -183,3 +187,19 @@ func (m *gitMetrics) recordLFSPhaseBytes(ctx context.Context, repo, phase string attribute.String("phase", phase), )) } + +// recordUploadPackCloneGate records the outcome of the clone-shaped +// upload-pack concurrency gate. Result is "admitted", "rejected", or "canceled". +func (m *gitMetrics) recordUploadPackCloneGate(ctx context.Context, result, repo string, waited time.Duration) { + if m == nil { + return + } + attrs := metric.WithAttributes( + attribute.String("result", result), + attribute.String("repository", repo), + ) + m.uploadPackCloneGateTotal.Add(ctx, 1, attrs) + if waited > 0 { + m.uploadPackCloneQueueWait.Record(ctx, waited.Seconds(), attrs) + } +} diff --git a/internal/strategy/git/uploadpack_limit.go b/internal/strategy/git/uploadpack_limit.go new file mode 100644 index 0000000..bc0f072 --- /dev/null +++ b/internal/strategy/git/uploadpack_limit.go @@ -0,0 +1,184 @@ +package git + +import ( + "context" + "net/http" + "strconv" + "sync" + "time" + + "github.com/block/cachew/internal/logging" +) + +type acquireResult int + +const ( + acquireOK acquireResult = iota + acquireOverloaded + acquireCanceled +) + +func (r acquireResult) String() string { + switch r { + case acquireOK: + return "admitted" + case acquireOverloaded: + return "rejected" + case acquireCanceled: + return "canceled" + default: + return "unknown" + } +} + +// uploadPackLimiter bounds concurrent clone-shaped git-upload-pack responses. +// Incremental fetches are never passed through this type. +type uploadPackLimiter struct { + global chan struct{} // nil if unlimited + perRepoLimit int + perRepo sync.Map // string -> chan struct{} + queueTimeout time.Duration + retryAfter time.Duration +} + +func newUploadPackLimiter(cfg Config) *uploadPackLimiter { + l := &uploadPackLimiter{ + perRepoLimit: cfg.UploadPackClonePerRepoConcurrency, + queueTimeout: cfg.UploadPackCloneQueueTimeout, + retryAfter: cfg.UploadPackCloneRetryAfter, + } + if l.retryAfter <= 0 { + l.retryAfter = 30 * time.Second + } + if cfg.UploadPackCloneConcurrency > 0 { + l.global = newTokenSem(cfg.UploadPackCloneConcurrency) + } + return l +} + +func newTokenSem(n int) chan struct{} { + ch := make(chan struct{}, n) + for range n { + ch <- struct{}{} + } + return ch +} + +func (l *uploadPackLimiter) enabled() bool { + return l != nil && (l.global != nil || l.perRepoLimit > 0) +} + +func (l *uploadPackLimiter) repoSem(repo string) chan struct{} { + if v, ok := l.perRepo.Load(repo); ok { + return v.(chan struct{}) + } + ch := newTokenSem(l.perRepoLimit) + actual, _ := l.perRepo.LoadOrStore(repo, ch) + return actual.(chan struct{}) +} + +func (l *uploadPackLimiter) acquire(ctx context.Context, repo string) (func(), acquireResult, time.Duration) { + release := func() {} + if !l.enabled() { + return release, acquireOK, 0 + } + start := time.Now() + var deadline time.Time + if l.queueTimeout > 0 { + deadline = start.Add(l.queueTimeout) + } + + if l.global != nil { + if result := waitToken(ctx, l.global, deadline); result != acquireOK { + return release, result, time.Since(start) + } + } + + var repoCh chan struct{} + if l.perRepoLimit > 0 { + repoCh = l.repoSem(repo) + if result := waitToken(ctx, repoCh, deadline); result != acquireOK { + if l.global != nil { + l.global <- struct{}{} + } + return release, result, time.Since(start) + } + } + + return func() { + if repoCh != nil { + repoCh <- struct{}{} + } + if l.global != nil { + l.global <- struct{}{} + } + }, acquireOK, time.Since(start) +} + +func waitToken(ctx context.Context, ch chan struct{}, deadline time.Time) acquireResult { + if ctx.Err() != nil { + return acquireCanceled + } + if deadline.IsZero() { + select { + case <-ch: + return acquireOK + default: + return acquireOverloaded + } + } + remaining := time.Until(deadline) + if remaining <= 0 { + select { + case <-ch: + return acquireOK + default: + return acquireOverloaded + } + } + timer := time.NewTimer(remaining) + defer timer.Stop() + select { + case <-ch: + return acquireOK + case <-ctx.Done(): + return acquireCanceled + case <-timer.C: + return acquireOverloaded + } +} + +func (s *Strategy) gateCloneUploadPack(w http.ResponseWriter, r *http.Request, repo string) (func(), bool) { + release, result, waited := s.uploadPackLimiter.acquire(r.Context(), repo) + s.metrics.recordUploadPackCloneGate(r.Context(), result.String(), repo, waited) + switch result { + case acquireOK: + if waited > 0 { + logging.FromContext(r.Context()).InfoContext(r.Context(), + "Admitted clone-shaped upload-pack after queue wait", + "upstream", repo, "waited", waited) + } + return release, true + case acquireCanceled: + return func() {}, false + case acquireOverloaded: + s.rejectCloneUploadPack(w, r, repo) + return func() {}, false + default: + s.rejectCloneUploadPack(w, r, repo) + return func() {}, false + } +} + +func (s *Strategy) rejectCloneUploadPack(w http.ResponseWriter, r *http.Request, repo string) { + retryAfter := 30 * time.Second + if s.uploadPackLimiter != nil && s.uploadPackLimiter.retryAfter > 0 { + retryAfter = s.uploadPackLimiter.retryAfter + } + sec := max(int(retryAfter.Seconds()), 1) + logging.FromContext(r.Context()).WarnContext(r.Context(), + "Rejecting clone-shaped upload-pack due to concurrency limit", + "upstream", repo, "retry_after", retryAfter) + w.Header().Set("Retry-After", strconv.Itoa(sec)) + http.Error(w, "too many concurrent git clones; retry later", http.StatusServiceUnavailable) +} diff --git a/internal/strategy/git/uploadpack_limit_http_test.go b/internal/strategy/git/uploadpack_limit_http_test.go new file mode 100644 index 0000000..274aeb0 --- /dev/null +++ b/internal/strategy/git/uploadpack_limit_http_test.go @@ -0,0 +1,84 @@ +package git_test + +import ( + "bytes" + "context" + "net/http" + "net/http/httptest" + "path/filepath" + "testing" + "time" + + "github.com/alecthomas/assert/v2" + + "github.com/block/cachew/internal/gitclone" + "github.com/block/cachew/internal/githubapp" + "github.com/block/cachew/internal/logging" + "github.com/block/cachew/internal/strategy/git" +) + +func TestCloneUploadPackRejectedBeforePackfile(t *testing.T) { + t.Parallel() + _, ctx := logging.Configure(context.Background(), logging.Config{}) + mux := newTestMux() + cm := gitclone.NewManagerProvider(ctx, gitclone.Config{ + MirrorRoot: filepath.Join(t.TempDir(), "clones"), + }, nil) + s, err := git.New(ctx, git.Config{ + UploadPackCloneConcurrency: 1, + UploadPackCloneRetryAfter: 15 * time.Second, + }, newTestScheduler(ctx, t), nil, mux, cm, func() (*githubapp.TokenManager, error) { return nil, nil }) //nolint:nilnil + assert.NoError(t, err) + waitForReady(t, s) + + release, admitted := s.HoldUploadPackCloneSlotForTest(ctx, "https://github.com/org/repo") + assert.True(t, admitted) + defer release() + + handler := mux.handlers["POST /git/{host}/{path...}"] + assert.NotZero(t, handler) + + req := httptest.NewRequestWithContext(ctx, http.MethodPost, + "/git/github.com/org/repo.git/git-upload-pack", http.NoBody) + req.SetPathValue("host", "github.com") + req.SetPathValue("path", "org/repo.git/git-upload-pack") + w := httptest.NewRecorder() + handler.ServeHTTP(w, req) + + assert.Equal(t, http.StatusServiceUnavailable, w.Code) + assert.Equal(t, "15", w.Header().Get("Retry-After")) + assert.Contains(t, w.Body.String(), "too many concurrent git clones") +} + +func TestIncrementalUploadPackNotGated(t *testing.T) { + t.Parallel() + _, ctx := logging.Configure(context.Background(), logging.Config{}) + mux := newTestMux() + cm := gitclone.NewManagerProvider(ctx, gitclone.Config{ + MirrorRoot: filepath.Join(t.TempDir(), "clones"), + }, nil) + s, err := git.New(ctx, git.Config{ + UploadPackCloneConcurrency: 1, + UploadPackCloneRetryAfter: 15 * time.Second, + }, newTestScheduler(ctx, t), nil, mux, cm, func() (*githubapp.TokenManager, error) { return nil, nil }) //nolint:nilnil + assert.NoError(t, err) + waitForReady(t, s) + + release, admitted := s.HoldUploadPackCloneSlotForTest(ctx, "https://127.0.0.1/org/other") + assert.True(t, admitted) + defer release() + + handler := mux.handlers["POST /git/{host}/{path...}"] + reqCtx, cancel := context.WithTimeout(ctx, 5*time.Second) + defer cancel() + req := httptest.NewRequestWithContext(reqCtx, http.MethodPost, + "/git/127.0.0.1/org/repo.git/git-upload-pack", + bytes.NewReader([]byte("have 0123456789abcdef0123456789abcdef01234567\n"))) + req.SetPathValue("host", "127.0.0.1") + req.SetPathValue("path", "org/repo.git/git-upload-pack") + w := httptest.NewRecorder() + handler.ServeHTTP(w, req) + + assert.NotEqual(t, http.StatusServiceUnavailable, w.Code) + assert.Equal(t, "", w.Header().Get("Retry-After")) +} diff --git a/internal/strategy/git/uploadpack_limit_test.go b/internal/strategy/git/uploadpack_limit_test.go new file mode 100644 index 0000000..a6cc213 --- /dev/null +++ b/internal/strategy/git/uploadpack_limit_test.go @@ -0,0 +1,160 @@ +package git //nolint:testpackage + +import ( + "context" + "sync" + "sync/atomic" + "testing" + "time" + + "github.com/alecthomas/assert/v2" +) + +func TestUploadPackLimiterUnlimited(t *testing.T) { + t.Parallel() + l := newUploadPackLimiter(Config{}) + var releases []func() + for range 32 { + release, result, waited := l.acquire(t.Context(), "https://github.com/org/repo") + assert.Equal(t, acquireOK, result) + assert.Equal(t, time.Duration(0), waited) + releases = append(releases, release) + } + for _, release := range releases { + release() + } +} + +func TestUploadPackLimiterGlobalRejectsWhenFull(t *testing.T) { + t.Parallel() + l := newUploadPackLimiter(Config{UploadPackCloneConcurrency: 1}) + release, result, _ := l.acquire(t.Context(), "https://github.com/org/a") + assert.Equal(t, acquireOK, result) + defer release() + + _, result, _ = l.acquire(t.Context(), "https://github.com/org/b") + assert.Equal(t, acquireOverloaded, result) +} + +func TestUploadPackLimiterPerRepoIndependent(t *testing.T) { + t.Parallel() + l := newUploadPackLimiter(Config{UploadPackClonePerRepoConcurrency: 1}) + releaseA, result, _ := l.acquire(t.Context(), "https://github.com/org/a") + assert.Equal(t, acquireOK, result) + defer releaseA() + + releaseB, result, _ := l.acquire(t.Context(), "https://github.com/org/b") + assert.Equal(t, acquireOK, result) + defer releaseB() + + _, result, _ = l.acquire(t.Context(), "https://github.com/org/a") + assert.Equal(t, acquireOverloaded, result) +} + +func TestUploadPackLimiterQueueThenAdmit(t *testing.T) { + t.Parallel() + l := newUploadPackLimiter(Config{ + UploadPackCloneConcurrency: 1, + UploadPackCloneQueueTimeout: time.Second, + }) + release, result, _ := l.acquire(t.Context(), "https://github.com/org/repo") + assert.Equal(t, acquireOK, result) + + var got acquireResult + var waited time.Duration + done := make(chan struct{}) + go func() { + defer close(done) + var queuedRelease func() + queuedRelease, got, waited = l.acquire(t.Context(), "https://github.com/org/other") + if got == acquireOK { + queuedRelease() + } + }() + + time.Sleep(50 * time.Millisecond) + release() + <-done + assert.Equal(t, acquireOK, got) + assert.True(t, waited >= 40*time.Millisecond) +} + +func TestUploadPackLimiterQueueTimeoutRejects(t *testing.T) { + t.Parallel() + l := newUploadPackLimiter(Config{ + UploadPackCloneConcurrency: 1, + UploadPackCloneQueueTimeout: 30 * time.Millisecond, + }) + release, result, _ := l.acquire(t.Context(), "https://github.com/org/repo") + assert.Equal(t, acquireOK, result) + defer release() + + start := time.Now() + _, result, _ = l.acquire(t.Context(), "https://github.com/org/other") + assert.Equal(t, acquireOverloaded, result) + assert.True(t, time.Since(start) >= 30*time.Millisecond) +} + +func TestUploadPackLimiterCanceled(t *testing.T) { + t.Parallel() + l := newUploadPackLimiter(Config{ + UploadPackCloneConcurrency: 1, + UploadPackCloneQueueTimeout: time.Second, + }) + release, result, _ := l.acquire(t.Context(), "https://github.com/org/repo") + assert.Equal(t, acquireOK, result) + defer release() + + ctx, cancel := context.WithCancel(t.Context()) + cancel() + _, result, _ = l.acquire(ctx, "https://github.com/org/other") + assert.Equal(t, acquireCanceled, result) +} + +func TestUploadPackLimiterReleasesGlobalIfPerRepoRejects(t *testing.T) { + t.Parallel() + l := newUploadPackLimiter(Config{ + UploadPackCloneConcurrency: 1, + UploadPackClonePerRepoConcurrency: 1, + }) + release, result, _ := l.acquire(t.Context(), "https://github.com/org/a") + assert.Equal(t, acquireOK, result) + + _, result, _ = l.acquire(t.Context(), "https://github.com/org/a") + assert.Equal(t, acquireOverloaded, result) + + release() + releaseB, result, _ := l.acquire(t.Context(), "https://github.com/org/b") + assert.Equal(t, acquireOK, result) + releaseB() +} + +func TestUploadPackLimiterConcurrentRespectsGlobal(t *testing.T) { + t.Parallel() + const limit = 3 + l := newUploadPackLimiter(Config{UploadPackCloneConcurrency: limit}) + var inFlight atomic.Int32 + var maxInFlight atomic.Int32 + var wg sync.WaitGroup + for range 20 { + wg.Go(func() { + release, result, _ := l.acquire(t.Context(), "https://github.com/org/repo") + if result != acquireOK { + return + } + n := inFlight.Add(1) + for { + cur := maxInFlight.Load() + if n <= cur || maxInFlight.CompareAndSwap(cur, n) { + break + } + } + time.Sleep(10 * time.Millisecond) + inFlight.Add(-1) + release() + }) + } + wg.Wait() + assert.True(t, maxInFlight.Load() <= limit) + assert.True(t, maxInFlight.Load() > 0) +}