feat(objgitd): bound the number of concurrent pushes

README.md

@@ -71,6 +71,14 @@ GIT_BIND=:9418
 HTTP_BIND=:8080
 SSH_BIND=:2222
 
+# Pushes that unpack a packfile at the same time. This is what bounds
+# memory under concurrent pushes: each one costs roughly 400 MiB of
+# resident set for a large repository. 0 disables the limit.
+MAX_CONCURRENT_PUSHES=4
+# How long a push waits for a slot before it fails,
+# in go time.Duration format
+PUSH_QUEUE_TIMEOUT=2m
+
 # Prometheus metrics
 METRICS_BIND=:9090
 

cmd/objgitd/git_protocol.go

@@ -53,6 +53,14 @@ type daemon struct {
 	// allowHooks gates running .objgit/hooks/receive-pack after a push.
 	allowHooks  bool
 	hookTimeout time.Duration
+
+	// pushes bounds how many pushes unpack a packfile at once, which is the
+	// only thing that bounds the daemon's resident set under concurrent pushes.
+	// A nil pushes is unlimited, which is what every test that does not care
+	// about the cap gets. Fetches are deliberately not gated: they allocate
+	// very differently, and one semaphore over both would let a burst of clones
+	// block pushes for reasons that have nothing to do with memory.
+	pushes *pushLimiter
 }
 
 // storerFor reports whether a repository already exists at st, returning st

cmd/objgitd/hooks.go

@@ -85,9 +85,13 @@ func diffRefs(before, after map[plumbing.ReferenceName]plumbing.Hash) []refUpdat
 // used for ref snapshots and hook checkouts — all three transports now share the
 // same Scanner-bounded PackfileWriter path (see writePack), so no transport needs
 // a capability-hiding wrapper.
+//
+// This is also the one place every push funnels through — smart HTTP and SSH
+// both land here, and git:// never serves receive-pack at all — so it is where
+// the push concurrency cap is applied, via the d.pushes.admit seam.
 func (d *daemon) receivePack(ctx context.Context, st storage.Storer, repoPath string, r io.ReadCloser, w io.WriteCloser, req *transport.ReceivePackRequest) error {
 	if !d.allowHooks {
-		err := receivePackStreaming(ctx, st, r, w, req, nil)
+		err := receivePackStreaming(ctx, st, r, w, req, d.pushes.admit, nil)
 		d.healHEADAfterPush(err, st, repoPath)
 		return err
 	}
@@ -114,7 +118,7 @@ func (d *daemon) receivePack(ctx context.Context, st storage.Storer, repoPath st
 		d.runHooks(repoPath, "receive-pack", st, updates, progress)
 	}
 
-	err = receivePackStreaming(ctx, st, r, w, req, onUpdated)
+	err = receivePackStreaming(ctx, st, r, w, req, d.pushes.admit, onUpdated)
 	d.healHEADAfterPush(err, st, repoPath)
 	return err
 }

cmd/objgitd/main.go

@@ -48,6 +48,9 @@ var (
 
 	packCompression = flag.Bool("pack-compression", true, "store zstd-compressed payloads in newly written pack containers; reading compressed containers is always enabled, so this is safe to turn off for one release before a rollback")
 	packedRefs      = flag.Bool("packed-refs", true, "write every ref into one packed-refs object under a compare-and-swap, instead of one object per ref; reading packed refs is always enabled, so this is safe to turn off for one release before a rollback")
+
+	maxConcurrentPushes = flag.Int("max-concurrent-pushes", 4, "pushes allowed to unpack a packfile at the same time; each one costs roughly 400 MiB of resident set for a large repository, so this is what bounds memory under concurrent pushes; 0 disables the limit")
+	pushQueueTimeout    = flag.Duration("push-queue-timeout", 2*time.Minute, "how long a push waits for a slot before it fails")
 )
 
 // tigrisBase adapts *tigris.Storer to repofs.Base: Storer.Scoped returns the
@@ -135,6 +138,7 @@ func main() {
 		authz:       auth.AllowAnonymous{AllowWrite: *allowPush},
 		allowHooks:  *allowHooks,
 		hookTimeout: *hookTimeout,
+		pushes:      newPushLimiter(*maxConcurrentPushes, *pushQueueTimeout),
 	}
 
 	slog.Info("objgitd listening",
@@ -146,6 +150,8 @@ func main() {
 		"allow_push", *allowPush,
 		"allow_hooks", *allowHooks,
 		"pack_cache_bytes", *packCacheBytes,
+		"max_concurrent_pushes", *maxConcurrentPushes,
+		"push_queue_timeout", *pushQueueTimeout,
 	)
 
 	g, gCtx := errgroup.WithContext(ctx)

cmd/objgitd/pushlimit.go

@@ -0,0 +1,103 @@
+package main
+
+import (
+	"context"
+	"errors"
+	"fmt"
+	"time"
+
+	"github.com/tigrisdata/objgit/internal/metrics"
+	"golang.org/x/sync/semaphore"
+)
+
+// errPushQueueTimeout is the sentinel behind every "waited too long for a push
+// slot" failure. It is wrapped, not returned directly, so the client-facing
+// message can name the deadline while tests and log readers can still match on
+// one error.
+var errPushQueueTimeout = errors.New("timed out waiting for a push slot")
+
+// admitFunc asks for permission to unpack a packfile. It blocks until a slot is
+// free, the deadline passes, or ctx is done, and returns the release closure for
+// the slot it acquired. A nil admitFunc means unlimited.
+//
+// The returned release must be called exactly once, and only when err is nil.
+type admitFunc func(ctx context.Context) (release func(), err error)
+
+// pushLimiter bounds how many pushes unpack a packfile at the same time.
+//
+// Memory during a push is dominated by the packfile the client is sending, and
+// it scales with the number of pushes in flight rather than with anything the
+// daemon controls: cmd/membench measured a flat ~429 MiB of resident set per
+// concurrent push of a 48 MiB pack, with no ceiling. GOMEMLIMIT halves that
+// slope but does not bound it, because it makes the collector work harder as
+// the heap grows instead of stopping the heap from growing. A count is the only
+// thing that turns "linear in whoever is pushing" into a number that can be
+// sized against a container.
+//
+// A push that arrives at a busy daemon waits rather than failing, because a git
+// client handles a slow push far better than a failed one, and a push that has
+// already uploaded its pack should not be thrown away because a peer was
+// mid-flight. Past the deadline it fails cleanly, which stops the queue from
+// growing without limit behind one slow push.
+type pushLimiter struct {
+	// sem is nil when the limit is disabled, which makes admit a no-op.
+	sem *semaphore.Weighted
+	// wait bounds how long a still-connected client queues for a slot. Zero
+	// means it waits as long as its context lives.
+	wait time.Duration
+}
+
+// newPushLimiter builds the limiter for max simultaneous pushes, each waiting
+// at most wait for a slot. max <= 0 disables the limit, matching how
+// -pack-cache-bytes treats 0 as "turn the feature off" rather than "zero
+// budget", so today's unbounded behavior stays reachable with one flag.
+func newPushLimiter(max int, wait time.Duration) *pushLimiter {
+	if max <= 0 {
+		return &pushLimiter{}
+	}
+	return &pushLimiter{sem: semaphore.NewWeighted(int64(max)), wait: wait}
+}
+
+// admit acquires one push slot. It is safe on a nil receiver, so a daemon built
+// without a limiter (every test that does not care) is unlimited.
+//
+// The wait deadline is layered on top of ctx rather than replacing it, so a
+// client that hangs up while queued gives up its place immediately instead of
+// holding it for the rest of the deadline.
+func (l *pushLimiter) admit(ctx context.Context) (func(), error) {
+	if l == nil || l.sem == nil {
+		return func() {}, nil
+	}
+
+	start := time.Now()
+	doneWaiting := metrics.TrackPushWait()
+
+	waitCtx := ctx
+	if l.wait > 0 {
+		var cancel context.CancelFunc
+		waitCtx, cancel = context.WithTimeout(ctx, l.wait)
+		defer cancel()
+	}
+
+	err := l.sem.Acquire(waitCtx, 1)
+	doneWaiting()
+
+	if err != nil {
+		// ctx outlives waitCtx, so a live ctx means the deadline expired and a
+		// dead one means the client left. The two are counted apart because
+		// only the first says the cap is too low.
+		if ctxErr := ctx.Err(); ctxErr != nil {
+			metrics.ObservePushWait(metrics.PushCanceled, start)
+			return nil, fmt.Errorf("objgitd: push abandoned while waiting for a slot: %w", ctxErr)
+		}
+		metrics.ObservePushWait(metrics.PushTimeout, start)
+		return nil, fmt.Errorf("objgitd: too many concurrent pushes, %w after %s", errPushQueueTimeout, l.wait)
+	}
+
+	metrics.ObservePushWait(metrics.PushAdmitted, start)
+	releaseSlot := metrics.TrackPushSlot()
+	return func() {
+		releaseSlot()
+		l.sem.Release(1)
+	}, nil
+}