feat(git): queue clone-shaped upload-pack under load #410
There are no files selected for viewing
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -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) | ||
|
Author
There was a problem hiding this comment. 🤖 M1 (major): the global token acquired at L92 is held while this waits for the per-repo slot, sharing the same deadline. A hot repo's waiters therefore pin global tokens: with 8/2/60s and 25 java clones, 6 java waiters occupy 6/8 global slots for 60s at a time while only 2 clones run, and go-square (2 free slots) is rejected. Repro (global=2, per-repo=1, timeout=2s; A running + A queued): B waited 1.9s and was admitted only when A's waiter timed out. Acquire per-repo first, then global — a per-repo waiter then blocks only its own repo, which needs a global slot anyway. |
||
| 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) | ||
|
Author
There was a problem hiding this comment. 🤖 m1 (minor): this records |
||
| 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) | ||
|
Author
There was a problem hiding this comment. 🤖 M2 (major): the PR description says "Git retries 503; clients should honor Retry-After" — upstream git does not. |
||
| } | ||
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
🤖 B1 (blocker for enabling in prod) + M3/M4:
RequestIsClonetreats any upload-pack POST withouthaveas a clone. Git's promisor lazy fetch runsgit -c fetch.negotiationAlgorithm=noop fetch … --filter=blob:none --stdin(upstreampromisor-remote.c:47-50), i.e. wants +filter blob:none+doneand no haves. Prodsnapshot-filtersappliesblob:nonetosquareup/java,ios-register,cash-ios, so every workstationgit log -p/blame/checkout <old>on those repos issues such POSTs; underper-repo = 2they compete with 200–760s full clones and will mostly 503, failing the user's git command. Repro: v2 bodycommand=fetch … want <oid> filter blob:none done→true. Also (M4) the detector fails closed when the 64 KiB inspect window is exhausted without seeinghave— 1500 wants + 1 have (75 KB) →true. Options: exempt bodies containing afilterline, or gate the expensive stage viauploadpack.packObjectsHook(sees the resolved want set), orcat-file --batch-checkthe wants and exempt non-commit wants; and fail open at the inspect limit.Placement (M3/m6): this runs before the
switch state, so it also gates — and 503s — thenot our reffallback and theStateEmpty/StateCloningspool/forward paths, which are GitHub-bound and cheap for Cachew, and a rejectedStateEmptyrequest never reaches thescheduler.Submit("clone")at L424. Production telemetry for the 2026-09-16 window shows Cachew emitted 0 POST 503s on/git(503s: 20, all GET; POST 502s: 66 =Upstream request failed (context canceled)×66, plusunexpected EOF×37 andFalling back to upstream due to 'not our ref'for java/go-square ×11) — i.e. the failures Cachew recorded were upstream-proxied fetches, not local pack generation. Worth verifying the root cause with server-side status metrics before tuning the HCL.