feat(git): add per-client clone rate limiting

Add a per-client-IP concurrent clone limiter to prevent a single client
from monopolizing all clone capacity. When a client exceeds the
configured limit, requests are rejected with 429 Too Many Requests
and a Retry-After header.

Configurable via max-clones-per-client in the git strategy config
block. Defaults to 0 (disabled). The tracker counts in-flight clones
per client IP and releases slots when clone jobs complete.

New metric cachew.git.clone_rejections_total (by client IP) provides
observability into rate-limited requests.

Co-authored-by: Amp <amp@ampcode.com>
Amp-Thread-ID: https://ampcode.com/threads/T-019d4a57-1477-707c-bb89-5543fddff0e7

Author: worstell

internal/strategy/git/git.go

@@ -39,8 +39,9 @@ type Config struct {
 	MirrorSnapshotInterval time.Duration `hcl:"mirror-snapshot-interval,optional" help:"How often to generate mirror snapshots for pod bootstrap. 0 uses snapshot-interval. Defaults to 2h." default:"2h"`
 	RepackInterval         time.Duration `hcl:"repack-interval,optional" help:"How often to run full repack. 0 disables." default:"0"`
 	// ServerURL is embedded as remote.origin.url in snapshots so git pull goes through cachew.
-	ServerURL   string `hcl:"server-url,optional" help:"Base URL of this cachew instance, embedded in snapshot remote URLs." default:"${CACHEW_URL}"`
-	ZstdThreads int    `hcl:"zstd-threads,optional" help:"Threads for zstd compression/decompression (0 = all CPU cores)." default:"0"`
+	ServerURL          string `hcl:"server-url,optional" help:"Base URL of this cachew instance, embedded in snapshot remote URLs." default:"${CACHEW_URL}"`
+	ZstdThreads        int    `hcl:"zstd-threads,optional" help:"Threads for zstd compression/decompression (0 = all CPU cores)." default:"0"`
+	MaxClonesPerClient int    `hcl:"max-clones-per-client,optional" help:"Max concurrent clone triggers per client IP. 0 disables per-client limiting." default:"0"`
 }
 
 type Strategy struct {
@@ -59,6 +60,7 @@ type Strategy struct {
 	coldSnapshotMu      sync.Map // keyed by upstream URL, values are *coldSnapshotEntry
 	deferredRestoreOnce sync.Map // keyed by upstream URL, ensures at most one deferred restore per repo
 	metrics             *gitMetrics
+	cloneTracker        *ClientCloneTracker // nil when per-client limiting is disabled
 }
 
 func New(
@@ -125,6 +127,9 @@ func New(
 		tokenManager: tokenManager,
 		metrics:      m,
 	}
+	if config.MaxClonesPerClient > 0 {
+		s.cloneTracker = NewClientCloneTracker(config.MaxClonesPerClient)
+	}
 	s.config.ServerURL = strings.TrimRight(config.ServerURL, "/")
 
 	if err := s.warmExistingRepos(ctx); err != nil {
@@ -274,10 +279,9 @@ func (s *Strategy) handleRequest(w http.ResponseWriter, r *http.Request) {
 
 	case gitclone.StateCloning, gitclone.StateEmpty:
 		if state == gitclone.StateEmpty {
-			logger.DebugContext(ctx, "Starting background clone, forwarding to upstream")
-			s.scheduler.Submit(repo.UpstreamURL(), "clone", func(ctx context.Context) error {
-				return s.startClone(ctx, repo)
-			})
+			if rejected := s.submitClone(w, r, repo); rejected {
+				return
+			}
 		}
 		if err := s.serveWithSpool(w, r, host, pathValue, upstreamURL); err != nil {
 			logger.WarnContext(ctx, "Spool failed, forwarding to upstream", "error", err)
@@ -286,6 +290,36 @@ func (s *Strategy) handleRequest(w http.ResponseWriter, r *http.Request) {
 	}
 }
 
+// submitClone submits a background clone job for the given repo, applying
+// per-client rate limiting when configured. Returns true if the request was
+// rejected (429 already written to w), false if the clone was submitted.
+func (s *Strategy) submitClone(w http.ResponseWriter, r *http.Request, repo *gitclone.Repository) bool {
+	ctx := r.Context()
+	logger := logging.FromContext(ctx)
+
+	if s.cloneTracker != nil {
+		ip := ClientIP(r)
+		release, ok := s.cloneTracker.TryAcquire(ip)
+		if !ok {
+			logger.WarnContext(ctx, "Per-client clone limit reached", "client", ip, "upstream", repo.UpstreamURL())
+			s.metrics.recordCloneRejection(ctx, ip)
+			w.Header().Set("Retry-After", "30")
+			http.Error(w, "Too many concurrent clone requests", http.StatusTooManyRequests)
+			return true
+		}
+		s.scheduler.Submit(repo.UpstreamURL(), "clone", func(ctx context.Context) error {
+			defer release()
+			return s.startClone(ctx, repo)
+		})
+	} else {
+		s.scheduler.Submit(repo.UpstreamURL(), "clone", func(ctx context.Context) error {
+			return s.startClone(ctx, repo)
+		})
+	}
+	logger.DebugContext(ctx, "Starting background clone, forwarding to upstream")
+	return false
+}
+
 func (s *Strategy) serveReadyRepo(w http.ResponseWriter, r *http.Request, repo *gitclone.Repository, host, pathValue string, isInfoRefs bool) error {
 	ctx := r.Context()
 

internal/strategy/git/metrics.go

@@ -11,9 +11,10 @@ import (
 )
 
 type gitMetrics struct {
-	operationDuration metric.Float64Histogram
-	operationTotal    metric.Int64Counter
-	requestTotal      metric.Int64Counter
+	operationDuration    metric.Float64Histogram
+	operationTotal       metric.Int64Counter
+	requestTotal         metric.Int64Counter
+	cloneRejectionsTotal metric.Int64Counter
 }
 
 func newGitMetrics() (*gitMetrics, error) {
@@ -39,6 +40,12 @@ func newGitMetrics() (*gitMetrics, error) {
 		return nil, errors.Wrap(err, "create requests_total counter")
 	}
 
+	if m.cloneRejectionsTotal, err = meter.Int64Counter("cachew.git.clone_rejections_total",
+		metric.WithDescription("Clone triggers rejected by per-client rate limiting"),
+		metric.WithUnit("{rejections}")); err != nil {
+		return nil, errors.Wrap(err, "create clone_rejections_total counter")
+	}
+
 	return m, nil
 }
 
@@ -55,3 +62,7 @@ func (m *gitMetrics) recordOperation(ctx context.Context, operation, status stri
 func (m *gitMetrics) recordRequest(ctx context.Context, requestType string) {
 	m.requestTotal.Add(ctx, 1, metric.WithAttributes(attribute.String("type", requestType)))
 }
+
+func (m *gitMetrics) recordCloneRejection(ctx context.Context, clientIP string) {
+	m.cloneRejectionsTotal.Add(ctx, 1, metric.WithAttributes(attribute.String("client", clientIP)))
+}

internal/strategy/git/ratelimit.go

@@ -0,0 +1,67 @@
+package git
+
+import (
+	"net"
+	"net/http"
+	"sync"
+	"sync/atomic"
+)
+
+// ClientCloneTracker limits how many concurrent clone operations a single
+// client IP can trigger. When a client exceeds the limit, the request is
+// rejected with 429 Too Many Requests.
+type ClientCloneTracker struct {
+	mu       sync.Mutex
+	inflight map[string]*atomic.Int32
+	limit    int
+}
+
+// NewClientCloneTracker creates a tracker that allows at most limit concurrent
+// clone operations per client IP.
+func NewClientCloneTracker(limit int) *ClientCloneTracker {
+	return &ClientCloneTracker{
+		inflight: make(map[string]*atomic.Int32),
+		limit:    limit,
+	}
+}
+
+// TryAcquire attempts to reserve a clone slot for the given client IP.
+// Returns true and a release function if the client is under the limit.
+// Returns false if the client has reached the limit.
+func (t *ClientCloneTracker) TryAcquire(clientIP string) (release func(), ok bool) {
+	t.mu.Lock()
+	counter, exists := t.inflight[clientIP]
+	if !exists {
+		counter = &atomic.Int32{}
+		t.inflight[clientIP] = counter
+	}
+	t.mu.Unlock()
+
+	if int(counter.Load()) >= t.limit {
+		return nil, false
+	}
+	counter.Add(1)
+
+	var once sync.Once
+	return func() {
+		once.Do(func() {
+			if counter.Add(-1) <= 0 {
+				t.mu.Lock()
+				// Re-check under lock to avoid racing with another acquire.
+				if counter.Load() <= 0 {
+					delete(t.inflight, clientIP)
+				}
+				t.mu.Unlock()
+			}
+		})
+	}, true
+}
+
+// ClientIP extracts the IP address from an HTTP request, stripping the port.
+func ClientIP(r *http.Request) string {
+	host, _, err := net.SplitHostPort(r.RemoteAddr)
+	if err != nil {
+		return r.RemoteAddr
+	}
+	return host
+}

internal/strategy/git/ratelimit_test.go

@@ -0,0 +1,100 @@
+package git_test
+
+import (
+	"net/http"
+	"sync"
+	"sync/atomic"
+	"testing"
+
+	"github.com/alecthomas/assert/v2"
+
+	"github.com/block/cachew/internal/strategy/git"
+)
+
+func TestClientCloneTrackerBasic(t *testing.T) {
+	tracker := git.NewClientCloneTracker(2)
+
+	r1, ok := tracker.TryAcquire("10.0.0.1")
+	assert.True(t, ok)
+
+	r2, ok := tracker.TryAcquire("10.0.0.1")
+	assert.True(t, ok)
+
+	// Third should be rejected.
+	_, ok = tracker.TryAcquire("10.0.0.1")
+	assert.False(t, ok)
+
+	// Different client should be fine.
+	r3, ok := tracker.TryAcquire("10.0.0.2")
+	assert.True(t, ok)
+
+	// Release one slot for 10.0.0.1.
+	r1()
+
+	r4, ok := tracker.TryAcquire("10.0.0.1")
+	assert.True(t, ok)
+
+	r2()
+	r3()
+	r4()
+}
+
+func TestClientCloneTrackerReleaseIdempotent(t *testing.T) {
+	tracker := git.NewClientCloneTracker(1)
+
+	release, ok := tracker.TryAcquire("10.0.0.1")
+	assert.True(t, ok)
+
+	// Double-release should not corrupt the counter.
+	release()
+	release()
+
+	r2, ok := tracker.TryAcquire("10.0.0.1")
+	assert.True(t, ok)
+	r2()
+}
+
+func TestClientCloneTrackerConcurrent(t *testing.T) {
+	tracker := git.NewClientCloneTracker(3)
+
+	// Hold 3 slots so all subsequent acquires must be rejected.
+	releases := make([]func(), 3)
+	for i := range 3 {
+		r, ok := tracker.TryAcquire("10.0.0.1")
+		assert.True(t, ok)
+		releases[i] = r
+	}
+
+	var rejected atomic.Int32
+	var wg sync.WaitGroup
+	for range 20 {
+		wg.Go(func() {
+			_, ok := tracker.TryAcquire("10.0.0.1")
+			if !ok {
+				rejected.Add(1)
+			}
+		})
+	}
+
+	wg.Wait()
+	assert.Equal(t, int32(20), rejected.Load(), "all attempts should be rejected while slots are held")
+
+	for _, r := range releases {
+		r()
+	}
+}
+
+func TestClientIP(t *testing.T) {
+	tests := []struct {
+		remoteAddr string
+		want       string
+	}{
+		{"192.168.1.1:12345", "192.168.1.1"},
+		{"[::1]:8080", "::1"},
+		{"192.168.1.1", "192.168.1.1"},
+	}
+	for _, tt := range tests {
+		r := &http.Request{RemoteAddr: tt.remoteAddr}
+		assert.Equal(t, tt.want, git.ClientIP(r), "ClientIP(%q)", tt.remoteAddr)
+	}
+}