From 326b6278f49c7025c7628a95e6f6cc61997df24e Mon Sep 17 00:00:00 2001 From: Xe Iaso Date: Mon, 31 Aug 2026 14:55:27 -0400 Subject: [PATCH] feat(objgitd): bound the number of concurrent pushes Memory during a push scales with the number of pushes in flight, and nothing in the daemon bounded that number. A sweep of concurrent pushes of a 48 MiB pack measured a flat 429 MiB of resident set for each one, with no ceiling: a dozen simultaneous pushes of a large repository is roughly 5 GB, and the process had no way to decline. GOMEMLIMIT halves the slope but bounds nothing, because it makes the collector work harder as the heap grows instead of stopping the heap from growing. Add a counting semaphore on *daemon. -max-concurrent-pushes sets the number of slots (default 4, 0 disables), and a push that finds them all taken waits up to -push-queue-timeout rather than failing at once, since a git client handles a slow push far better than a failed one. The gate sits inside receivePackStreaming, right after the capability decode and before the packfile is read. That is the one place both push transports reach, and the first point at which the response framing is known, so a push that gives up waiting is reported through sendReportStatus and the person pushing sees why instead of getting a dropped connection. The sideband setup moved above the packfile read to make that possible; nothing is written to the response in between, so the reorder is not observable on the wire. 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. Four objgit_push_* series make the queue visible, which is what turns "the daemon dies" into an improvement rather than a trade. objgit_push_queue_waiting is the one to alert on. Signed-off-by: Xe Iaso Assisted-by: Claude Opus 5 via Claude Code --- README.md | 8 + cmd/objgitd/git_protocol.go | 8 + cmd/objgitd/hooks.go | 8 +- cmd/objgitd/main.go | 6 + cmd/objgitd/pushlimit.go | 103 ++++++++ cmd/objgitd/pushlimit_test.go | 344 ++++++++++++++++++++++++++ cmd/objgitd/receivepack.go | 83 +++++-- docs/architecture/metrics.md | 40 ++- docs/architecture/transports.md | 39 ++- docs/plans/bound-concurrent-pushes.md | 184 ++++++++++++++ internal/metrics/metrics.go | 59 +++++ 11 files changed, 840 insertions(+), 42 deletions(-) create mode 100644 cmd/objgitd/pushlimit.go create mode 100644 cmd/objgitd/pushlimit_test.go create mode 100644 docs/plans/bound-concurrent-pushes.md diff --git a/README.md b/README.md index f715c5c..db34bef 100644 --- a/README.md +++ b/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 diff --git a/cmd/objgitd/git_protocol.go b/cmd/objgitd/git_protocol.go index 47812ef..68d560c 100644 --- a/cmd/objgitd/git_protocol.go +++ b/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 diff --git a/cmd/objgitd/hooks.go b/cmd/objgitd/hooks.go index 51e7cef..8c818f4 100644 --- a/cmd/objgitd/hooks.go +++ b/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 } diff --git a/cmd/objgitd/main.go b/cmd/objgitd/main.go index 2d62121..ec36ffb 100644 --- a/cmd/objgitd/main.go +++ b/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) diff --git a/cmd/objgitd/pushlimit.go b/cmd/objgitd/pushlimit.go new file mode 100644 index 0000000..5f05463 --- /dev/null +++ b/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 +} diff --git a/cmd/objgitd/pushlimit_test.go b/cmd/objgitd/pushlimit_test.go new file mode 100644 index 0000000..1c6863a --- /dev/null +++ b/cmd/objgitd/pushlimit_test.go @@ -0,0 +1,344 @@ +package main + +import ( + "context" + "errors" + "fmt" + "net" + "net/http/httptest" + "os/exec" + "strings" + "sync" + "testing" + "time" + + "github.com/go-git/go-billy/v6/memfs" + "github.com/prometheus/client_golang/prometheus" + "github.com/tigrisdata/objgit/internal/auth" + "github.com/tigrisdata/objgit/internal/repofs" +) + +// TestPushLimiterAdmit covers admission without a git client, so every case is +// decided by the limiter rather than by how fast a subprocess happens to run. +func TestPushLimiterAdmit(t *testing.T) { + for _, tt := range []struct { + name string + max int + wait time.Duration + held int // slots taken before the admit under test + hangUp bool // cancel the caller's context before it admits + wantErr error + }{ + { + name: "zero disables the cap", + max: 0, + wait: 50 * time.Millisecond, + held: 8, + }, + { + name: "admits below the cap", + max: 2, + wait: 50 * time.Millisecond, + held: 1, + }, + { + name: "gives up at the deadline once full", + max: 1, + wait: 50 * time.Millisecond, + held: 1, + wantErr: errPushQueueTimeout, + }, + { + name: "a client that hangs up does not wait out the deadline", + max: 1, + wait: time.Hour, + held: 1, + hangUp: true, + wantErr: context.Canceled, + }, + } { + t.Run(tt.name, func(t *testing.T) { + l := newPushLimiter(tt.max, tt.wait) + for i := range tt.held { + release, err := l.admit(context.Background()) + if err != nil { + t.Fatalf("taking slot %d: %v", i, err) + } + t.Cleanup(release) + } + + ctx, cancel := context.WithCancel(context.Background()) + defer cancel() + if tt.hangUp { + cancel() + } + + release, err := l.admit(ctx) + if tt.wantErr != nil { + if !errors.Is(err, tt.wantErr) { + t.Fatalf("admit err = %v, want one wrapping %v", err, tt.wantErr) + } + if release != nil { + t.Error("admit handed back a release alongside an error") + } + return + } + if err != nil { + t.Fatalf("admit: %v", err) + } + release() + }) + } +} + +// TestPushLimiterQueuedClientReleasesItsPlace covers the leak that would be +// silent until the daemon wedged: a push that queues and then loses its client +// must give up its place immediately. It is verified the only way it can be — +// by a later push taking the freed slot instead of waiting out a deadline it +// would never reach in a test. +func TestPushLimiterQueuedClientReleasesItsPlace(t *testing.T) { + l := newPushLimiter(1, time.Hour) + + held, err := l.admit(context.Background()) + if err != nil { + t.Fatalf("taking the only slot: %v", err) + } + + // A second push queues behind it, then its client hangs up. + queuedCtx, hangUp := context.WithCancel(context.Background()) + queued := make(chan error, 1) + go func() { + release, err := l.admit(queuedCtx) + if release != nil { + release() + } + queued <- err + }() + waitFor(t, "a push to start queueing", func() bool { + return gaugeValue(t, "objgit_push_queue_waiting") == 1 + }) + hangUp() + if err := <-queued; !errors.Is(err, context.Canceled) { + t.Fatalf("queued push err = %v, want one wrapping context.Canceled", err) + } + + // Hand the slot back. It must reach the third push, not the dead waiter. + held() + third := make(chan error, 1) + go func() { + release, err := l.admit(context.Background()) + if release != nil { + release() + } + third <- err + }() + select { + case err := <-third: + if err != nil { + t.Fatalf("third push: %v", err) + } + case <-time.After(30 * time.Second): + t.Fatal("third push never admitted; the abandoned push kept its place") + } + + assertPushGaugesDrained(t) +} + +// TestPushCapReportsTimeoutToClient drives a real git client at a daemon whose +// one push slot is already taken, so the push has to queue and then give up. It +// asserts the reason reaches the person pushing as a push failure rather than as +// a dropped connection, and that the gate is not sticky afterwards. +func TestPushCapReportsTimeoutToClient(t *testing.T) { + if _, err := exec.LookPath("git"); err != nil { + t.Skip("git not installed") + } + + d := &daemon{ + sysFS: memfs.New(), + resolver: repofs.BucketResolver{Base: newMemBase()}, + authz: auth.AllowAnonymous{AllowWrite: true}, + pushes: newPushLimiter(1, 250*time.Millisecond), + } + ts := httptest.NewServer(d.httpHandler()) + t.Cleanup(ts.Close) + + held, err := d.pushes.admit(context.Background()) + if err != nil { + t.Fatalf("taking the only push slot: %v", err) + } + + work := seedRepo(t) + remote := ts.URL + "/acme/queued.git" + + out, err := tryGit(work, "push", remote, "main") + if err == nil { + t.Fatalf("push should have failed while the only slot was held:\n%s", out) + } + for _, want := range []string{"too many concurrent pushes", "timed out waiting for a push slot"} { + if !strings.Contains(out, want) { + t.Errorf("push output does not explain the failure, missing %q:\n%s", want, out) + } + } + + held() + if out, err := tryGit(work, "push", remote, "main"); err != nil { + t.Fatalf("push after the slot was freed: %v\n%s", err, out) + } + + // Both the happy path and the timeout path above must have given their + // counters back. + assertPushGaugesDrained(t) +} + +// TestPushCapReportsTimeoutOverSSH is the same assertion over SSH, which is not +// the same code path: SSH is not a stateless RPC, so the ref advertisement goes +// out from inside receivePackStreaming before the gate is reached, and the +// sideband is framed on a long-lived connection instead of an HTTP response. +func TestPushCapReportsTimeoutOverSSH(t *testing.T) { + checkSSHBinaries(t) + + d := &daemon{ + sysFS: memfs.New(), + resolver: repofs.BucketResolver{Base: newMemBase()}, + authz: auth.AllowAnonymous{AllowWrite: true}, + pushes: newPushLimiter(1, 250*time.Millisecond), + } + srv, err := newSSHServer(d, "") + if err != nil { + t.Fatalf("newSSHServer: %v", err) + } + ln, err := net.Listen("tcp", "127.0.0.1:0") + if err != nil { + t.Fatalf("listen: %v", err) + } + go srv.Serve(ln) //nolint:errcheck // returns when ln closes + t.Cleanup(func() { srv.Close(); ln.Close() }) + + held, err := d.pushes.admit(context.Background()) + if err != nil { + t.Fatalf("taking the only push slot: %v", err) + } + + env := gitSSHEnv(t) + work := seedRepo(t) + remote := fmt.Sprintf("ssh://git@%s/acme/queued.git", ln.Addr().String()) + + out, err := gitWithEnv(work, env, "push", remote, "main") + if err == nil { + t.Fatalf("push should have failed while the only slot was held:\n%s", out) + } + if !strings.Contains(out, "too many concurrent pushes") { + t.Errorf("push output does not explain the failure:\n%s", out) + } + + held() + if out, err := gitWithEnv(work, env, "push", remote, "main"); err != nil { + t.Fatalf("push after the slot was freed: %v\n%s", err, out) + } + + assertPushGaugesDrained(t) +} + +// TestPushCapQueuesRatherThanFails pushes from several clients at once. Whatever +// the cap, every push must land: the cap is meant to turn a memory problem into +// a latency one, not into failures. +func TestPushCapQueuesRatherThanFails(t *testing.T) { + if _, err := exec.LookPath("git"); err != nil { + t.Skip("git not installed") + } + + const pushes = 4 + + for _, tt := range []struct { + name string + max int + }{ + {name: "one at a time", max: 1}, + {name: "two at a time", max: 2}, + {name: "zero disables the cap", max: 0}, + } { + t.Run(tt.name, func(t *testing.T) { + mb := newMemBase() + ts := httptest.NewServer((&daemon{ + sysFS: memfs.New(), + resolver: repofs.BucketResolver{Base: mb}, + authz: auth.AllowAnonymous{AllowWrite: true}, + pushes: newPushLimiter(tt.max, 60*time.Second), + }).httpHandler()) + t.Cleanup(ts.Close) + + works := make([]string, pushes) + for i := range works { + works[i] = seedRepo(t) + } + + var wg sync.WaitGroup + outs := make([]string, pushes) + errs := make([]error, pushes) + for i := range pushes { + wg.Go(func() { + remote := fmt.Sprintf("%s/acme/repo%d.git", ts.URL, i) + outs[i], errs[i] = tryGit(works[i], "push", remote, "main") + }) + } + wg.Wait() + + for i := range pushes { + if errs[i] != nil { + t.Errorf("push %d failed: %v\n%s", i, errs[i], outs[i]) + } + if repo := fmt.Sprintf("acme/repo%d", i); !mb.exists(repo) { + t.Errorf("push %d did not land: %s does not exist", i, repo) + } + } + + assertPushGaugesDrained(t) + }) + } +} + +// assertPushGaugesDrained checks both push gauges are back at zero. A semaphore +// released only on the happy path is the classic bug in this shape of change, +// and it is invisible from the outside until the daemon stops taking pushes. +func assertPushGaugesDrained(t *testing.T) { + t.Helper() + for _, name := range []string{"objgit_push_slots_held", "objgit_push_queue_waiting"} { + if got := gaugeValue(t, name); got != 0 { + t.Errorf("%s = %v after every push finished, want 0", name, got) + } + } +} + +// gaugeValue reads one gauge out of the default registry by name. The metrics +// package keeps its collectors unexported, and gathering by name avoids handing +// tests a hook into them. +func gaugeValue(t *testing.T, name string) float64 { + t.Helper() + families, err := prometheus.DefaultGatherer.Gather() + if err != nil { + t.Fatalf("gathering metrics: %v", err) + } + for _, f := range families { + if f.GetName() != name { + continue + } + for _, m := range f.GetMetric() { + return m.GetGauge().GetValue() + } + } + t.Fatalf("gauge %q not registered", name) + return 0 +} + +// waitFor polls cond until it holds, so a test can wait on a goroutine reaching +// a state instead of sleeping long enough to hope it did. +func waitFor(t *testing.T, what string, cond func() bool) { + t.Helper() + deadline := time.Now().Add(30 * time.Second) + for !cond() { + if time.Now().After(deadline) { + t.Fatalf("timed out waiting for %s", what) + } + time.Sleep(time.Millisecond) + } +} diff --git a/cmd/objgitd/receivepack.go b/cmd/objgitd/receivepack.go index 831c89e..dcb8cf8 100644 --- a/cmd/objgitd/receivepack.go +++ b/cmd/objgitd/receivepack.go @@ -22,22 +22,36 @@ import ( ) // receivePackStreaming is a fork of go-git's transport.ReceivePack (v6) that -// adds one seam: onUpdated runs after refs are updated and report-status is -// sent, but *before* the closing sideband flush-pkt. go-git keeps its sideband -// Muxer internal and flushes it before returning, so there is no other way to -// stream "remote:" progress to the client. The seam hands back a band-2 -// (sideband.ProgressMessage) writer when the client negotiated sideband, or nil -// otherwise — callers stream hook output through it and fall back to logging -// when it is nil. +// adds two seams: // -// Everything except the onUpdated seam mirrors transport.ReceivePack verbatim, -// including the helper functions copied below. +// - admit gates the memory-heavy part of the push. It runs once the client's +// capabilities are known and before the packfile is read, and the slot it +// returns is held until this function returns, which covers unpacking, the +// ref update, and any hooks. A nil admit means unlimited. +// - onUpdated runs after refs are updated and report-status is sent, but +// *before* the closing sideband flush-pkt. go-git keeps its sideband Muxer +// internal and flushes it before returning, so there is no other way to +// stream "remote:" progress to the client. The seam hands back a band-2 +// (sideband.ProgressMessage) writer when the client negotiated sideband, or +// nil otherwise — callers stream hook output through it and fall back to +// logging when it is nil. +// +// admit deliberately sits after the capability decode rather than at the top of +// the call: that is the first point at which the response framing is known, so a +// push that gives up waiting can be reported as a push failure the client +// renders ("error: remote unpack failed: ...") instead of a dropped connection. +// +// Everything except those two seams mirrors transport.ReceivePack verbatim, +// including the helper functions copied below. The sideband setup moved above +// the packfile read so the admit failure can use it; nothing is written to w in +// between, so the reorder is not observable on the wire. func receivePackStreaming( ctx context.Context, st storage.Storer, r io.ReadCloser, w io.WriteCloser, opts *transport.ReceivePackRequest, + admit admitFunc, onUpdated func(progress io.Writer), ) error { if w == nil { @@ -114,23 +128,6 @@ func receivePackStreaming( } } - // Receive the packfile - var unpackErr error - if needPackfile { - unpackErr = writePack(st, rd) - } - - // Done with the request, now close the reader - // to indicate that we are done reading from it. - if err := r.Close(); err != nil { - return fmt.Errorf("closing reader: %w", err) - } - - // Report status if the client supports it - if !updreq.Capabilities.Supports(capability.ReportStatus) { - return unpackErr - } - var ( useSideband bool mux *sideband.Muxer @@ -147,8 +144,40 @@ func receivePackStreaming( useSideband = true } } - writeCloser := ioutil.NewWriteCloser(writer, w) + reportStatus := caps.Supports(capability.ReportStatus) + + // Claim a push slot before reading the pack, and hold it until this call + // returns: the unpack, the ref update, and the hooks all allocate. + if admit != nil { + release, err := admit(ctx) + if err != nil { + if reportStatus { + _ = sendReportStatus(writeCloser, err, nil) + } + _ = closeWriter(w) + return err + } + defer release() + } + + // Receive the packfile + var unpackErr error + if needPackfile { + unpackErr = writePack(st, rd) + } + + // Done with the request, now close the reader + // to indicate that we are done reading from it. + if err := r.Close(); err != nil { + return fmt.Errorf("closing reader: %w", err) + } + + // Report status if the client supports it + if !reportStatus { + return unpackErr + } + if unpackErr != nil { res := sendReportStatus(writeCloser, unpackErr, nil) _ = closeWriter(w) diff --git a/docs/architecture/metrics.md b/docs/architecture/metrics.md index 6fa9f1c..cbea291 100644 --- a/docs/architecture/metrics.md +++ b/docs/architecture/metrics.md @@ -17,14 +17,38 @@ repository names have unbounded cardinality. Git operations are keyed by The package exposes thin helpers, so no call site carries label plumbing: -| Helper | Use | -| --------------- | -------------------------------------- | -| `ObserveS3` | The s3fs observer. | -| `ObserveGitOp` | One git operation. | -| `TrackInFlight` | Returns a deferred decrement. | -| `ObserveAuth` | Maps the `auth` enums to labels. | -| `ObserveHook` | One hook run. | -| `ReposCreated` | A new repository. | +| Helper | Use | +| ----------------- | -------------------------------- | +| `ObserveS3` | The s3fs observer. | +| `ObserveGitOp` | One git operation. | +| `TrackInFlight` | Returns a deferred decrement. | +| `ObserveAuth` | Maps the `auth` enums to labels. | +| `ObserveHook` | One hook run. | +| `ReposCreated` | A new repository. | +| `TrackPushWait` | Returns a deferred decrement. | +| `TrackPushSlot` | Returns a deferred decrement. | +| `ObservePushWait` | One wait for a push slot. | + +## The push queue + +`-max-concurrent-pushes` changes the failure mode under load. The daemon no +longer grows its heap without limit. It makes pushes queue instead. This is an +improvement only when the queue is visible, so the cap comes with four series: + +| Series | Type | Meaning | +| ---------------------------------- | --------- | ---------------------------------------- | +| `objgit_push_slots_held` | gauge | Pushes that unpack a packfile right now. | +| `objgit_push_queue_waiting` | gauge | Pushes that wait for a slot. | +| `objgit_push_queue_wait_seconds` | histogram | Time one push spent in the queue. | +| `objgit_push_queue_outcomes_total` | counter | Slot requests by outcome. | + +`objgit_push_queue_waiting` is the one to alert on. A value that stays above +zero means the cap is below the offered load. There is no other way to see +that from outside the process. + +The outcome label separates the two ways a wait ends badly. `timeout` means the +client waited out `-push-queue-timeout`, which is the signal that the cap is too +low. `canceled` means the client hung up while queued, which is not. ## Three instrumentation seams diff --git a/docs/architecture/transports.md b/docs/architecture/transports.md index 52d5952..600080b 100644 --- a/docs/architecture/transports.md +++ b/docs/architecture/transports.md @@ -4,11 +4,11 @@ objgitd speaks three git transports. All three answer the protocol natively with the same go-git `transport.*` functions. objgitd never runs the `git` binary, and it never writes a checkout to disk. -| Transport | Flag | Default | Credential | -| ----------- | -------------- | -------- | --------------------------- | -| Smart HTTP | `-http-bind` | `:8080` | HTTP Basic, or anonymous. | -| git:// | `-git-bind` | `:9418` | Anonymous only. | -| SSH | `-ssh-bind` | off | Public key, or anonymous. | +| Transport | Flag | Default | Credential | +| ---------- | ------------ | ------- | ------------------------- | +| Smart HTTP | `-http-bind` | `:8080` | HTTP Basic, or anonymous. | +| git:// | `-git-bind` | `:9418` | Anonymous only. | +| SSH | `-ssh-bind` | off | Public key, or anonymous. | All three route their decisions through [the auth seam](auth.md). @@ -124,3 +124,32 @@ and it reports its own error. A storer with no `refUpdater` keeps the per-reference path, which is what `memory.Storage` uses in the tests. + +## The push concurrency cap + +Memory during a push scales with the number of pushes in flight, and nothing in +the daemon bounds that number. A sweep of concurrent pushes of a 48 MiB pack +measured a flat 429 MiB of resident set for each one, with no ceiling. +`GOMEMLIMIT` halves the slope. It does not bound anything, because it makes the +collector work harder as the heap grows instead of stopping the heap from +growing. + +`pushlimit.go` holds a counting semaphore on `*daemon`. `-max-concurrent-pushes` +sets the number of slots, and `0` disables the limit. A push that finds every +slot taken waits, up to `-push-queue-timeout`, and then fails. + +`(*daemon).receivePack` is the one place to gate. Smart HTTP and SSH both reach +it, and git:// never serves `receive-pack` at all. 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. + +The gate itself sits a few lines further in, inside `receivePackStreaming`, +right after the client's capabilities are decoded and before the packfile is +read. That is the first point at which the response framing is known, so a push +that gives up waiting is reported through `sendReportStatus` and the person +pushing sees `error: remote unpack failed: objgitd: too many concurrent +pushes...`. A gate at the top of `receivePack` could only drop the connection. + +The slot is released by one `defer` that covers the unpack, the reference +update, and the hooks. Watch `objgit_push_queue_waiting`: see +[metrics.md](metrics.md). diff --git a/docs/plans/bound-concurrent-pushes.md b/docs/plans/bound-concurrent-pushes.md new file mode 100644 index 0000000..3cdd833 --- /dev/null +++ b/docs/plans/bound-concurrent-pushes.md @@ -0,0 +1,184 @@ +# Plan: bound the number of concurrent pushes + +## Context + +`cmd/membench` swept concurrent pushes of a 48 MiB pack against one daemon. +Memory is linear in concurrency, with a flat slope and no ceiling of its own: + +| K simultaneous pushes | Rise in RSS | Per concurrent push | Failures | +| --------------------- | ----------- | ------------------- | -------- | +| 2 | +814 MiB | 407 MiB | 0 | +| 4 | +1611 MiB | 403 MiB | 0 | +| 8 | +3431 MiB | 429 MiB | 0 | + +Nothing in `objgitd` bounds K. Whoever is pushing sets it. A dozen simultaneous +pushes of a large repository is roughly 5 GB of resident set, and the process +has no way to decline. + +`GOMEMLIMIT` halves the slope. It measured 215 MiB for each push instead of +429 MiB, with no failures. It does not bound anything either. It makes the +collector work harder as the heap grows. It does not stop the heap from growing. +Under enough concurrency the daemon still gets killed, only later. A limit is the +only thing that turns "linear forever" into a number that can be sized against. + +## Where the choke point is + +There is exactly one, and it already exists. `(*daemon).receivePack` +(`cmd/objgitd/hooks.go`) is reached from both transports that accept pushes: + +- Smart HTTP: `cmd/objgitd/http.go` +- SSH: `cmd/objgitd/ssh.go` + +The git:// server does not serve `receive-pack` at all. +`cmd/objgitd/git_protocol.go` only maps the service to a write operation for +authorization. One gate inside `(*daemon).receivePack` therefore covers every +push the daemon can take, and no transport needs to know about it. + +Fetches are deliberately out of scope. 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. + +## Decisions + +- **A counting semaphore on `*daemon`, taken in `(*daemon).receivePack` and + released on return.** One gate, at the one place both transports already + funnel through. `golang.org/x/sync/semaphore` comes with the + `golang.org/x/sync` module that the repo already requires for `errgroup`. + +- **Waiting, not rejecting, up to a deadline.** A push that arrives at a busy + daemon waits for a slot rather than failing at once. Git clients handle a slow + push far better than a failed one, and a push that has already uploaded its + pack must not be thrown away because a peer was mid-flight. Past the deadline + it fails cleanly. + +- **The deadline comes from the request context, plus its own flag.** The + semaphore acquire takes `ctx`, so a client that hangs up while queued gives up + its place at once. A separate `-push-queue-timeout` bounds how long a + still-connected client waits, so the queue cannot grow without limit behind + one slow push. + +- **Two flags, kebab-case with a `flagenv` fallback, per the repo convention:** + + | Flag | Env | Default | Meaning | + | ------------------------ | ----------------------- | ------- | ------------------------------------------------- | + | `-max-concurrent-pushes` | `MAX_CONCURRENT_PUSHES` | `4` | Pushes admitted at once. `0` disables the limit. | + | `-push-queue-timeout` | `PUSH_QUEUE_TIMEOUT` | `2m` | How long a push waits for a slot before it fails. | + + The default of 4 is not arbitrary. At the measured 429 MiB for each push it + puts the steady-state push ceiling near 1.7 GB, which fits a 2 GB container + once the encoder floor is removed. Set it together with that cap and with + `GOMEMLIMIT`. + +- **`0` means unlimited.** This matches how `-pack-cache-bytes` treats `0` as + "disable" rather than "zero budget". Anyone who wants today's behavior gets it + with one flag. + +- **The error reaches the client as a push failure with a readable message**, + through the existing report-status path (`sendReportStatus` in + `cmd/objgitd/receivepack.go`), so the person pushing sees why instead of + getting a dropped connection. + +## Where the gate landed, and why not at the top + +The two decisions above pull against each other. A gate at the very top of +`(*daemon).receivePack` runs before the client's capabilities are decoded, so at +that point the daemon does not know whether the response is plain pkt-line or +multiplexed on a sideband. It cannot write a report-status the client will parse. +All it can do is drop the connection, and over SSH it would also stall before +the reference advertisement. + +The gate therefore sits a few lines further in, as an `admitFunc` seam that +`(*daemon).receivePack` passes into `receivePackStreaming`. It fires right after +the capability decode and before the packfile is read. This keeps every property +the plan asked for: + +- One gate. Both push transports reach it, and neither knows about it. +- The slot covers the whole memory-heavy region: the unpack, the reference + update, and the hooks. +- A push that gives up waiting is reported the same way an unpack failure is. + git prints it as `error: remote unpack failed:` followed by the reason. + +The sideband setup in `receivePackStreaming` moved above the packfile read to +make that reporting possible. Nothing is written to the response between the old +position and the new one, so the reorder is not observable on the wire. + +## Observability + +This changes the failure mode from "the daemon dies" to "pushes queue", which is +an improvement only when the queue is visible. Four series, all under the +`objgit_push_` prefix: + +| Series | Type | Meaning | +| ---------------------------------- | --------- | ------------------------- | +| `objgit_push_slots_held` | gauge | Pushes that hold a slot. | +| `objgit_push_queue_waiting` | gauge | Pushes that wait for one. | +| `objgit_push_queue_wait_seconds` | histogram | Time spent waiting. | +| `objgit_push_queue_outcomes_total` | counter | Slot requests by outcome. | + +`objgit_push_queue_waiting` is the one that matters. A value that stays above +zero means the cap is below the offered load, and there is no other way to tell +that from outside the process. + +The outcome label separates the two ways a wait ends badly. `timeout` means the +client waited out `-push-queue-timeout`. `canceled` means it hung up while +queued. Only the first says the cap is too low. + +## Testing + +`cmd/objgitd/pushlimit_test.go`. The limiter's own semantics are tested without +a git client, so no case is decided by how fast a subprocess happens to run: + +- **`TestPushLimiterAdmit`.** A table over the cap, the deadline, and the number + of slots already taken: `0` disables the cap, a push below the cap is admitted, + a full house gives up at the deadline with an error wrapping + `errPushQueueTimeout`, and a client that hangs up does not wait out an hour. +- **`TestPushLimiterQueuedClientReleasesItsPlace`.** A push queues, its client + hangs up, and the held slot is then handed back. A third push must take that + slot. If the abandoned push kept its place, the third one waits forever. + +Then the wiring, with a real `git` client. Each test takes the daemon's only +slot from the test itself, which makes the queue deterministic: + +- **`TestPushCapReportsTimeoutToClient`.** Over smart HTTP, a push that finds the + slot held fails, and its output names the reason. The same push lands once the + slot is freed, so the gate is not sticky. +- **`TestPushCapReportsTimeoutOverSSH`.** The same assertion over SSH, which is + a different path: SSH is not a stateless RPC, so the reference advertisement + goes out from inside `receivePackStreaming` before the gate is reached. +- **`TestPushCapQueuesRatherThanFails`.** Four clients push at once against caps + of 1, 2, and 0. Every push must land. The cap turns a memory problem into a + latency problem, not into failures. + +Every one of these ends with `assertPushGaugesDrained`. A semaphore released +only on the happy path is the classic bug in this shape of change, and it is +invisible from the outside until the daemon stops taking pushes. + +## Risks + +- **The release must cover every return path**, including hook failures and + panics. One `defer` at the acquire point is the only acceptable form. +- **A cap below the number of pushes a CI system fires at once turns a memory + problem into a latency problem.** That is the intended trade, but it must be a + deliberate one. This is why the wait-queue gauge is not optional. +- **Interaction with the zstd encoder cap.** A push cap of 4 matches an encoder + cap of 4. If the push cap is later raised alone, compression becomes the + bottleneck and pushes get slower for a reason the push metrics do not show. +- **This does not reduce the cost of one push.** A lone push of a very large + repository still allocates whatever it allocates. + +## How this is verified + +Re-run the sweep and confirm the slope for each push stops mattering above the +cap: + +```sh +go run ./cmd/membench -conc-steps 1,2,4,8,16 +``` + +With `-max-concurrent-pushes 4`, peak RSS at K=8 and K=16 must land close to +peak RSS at K=4 instead of climbing. The failure column must stay at zero, and +the extra pushes must show up as longer wall clock. + +**`cmd/membench` is not in this repository.** This step is therefore not yet +run. The tests above cover admission, the deadline, the release, and the +client-visible failure. They do not measure resident set. diff --git a/internal/metrics/metrics.go b/internal/metrics/metrics.go index 334123e..1ac4e50 100644 --- a/internal/metrics/metrics.go +++ b/internal/metrics/metrics.go @@ -115,6 +115,35 @@ var ( Buckets: []float64{0.05, 0.1, 0.2, 0.3, 0.4, 0.5, 0.6, 0.7, 0.8, 0.9, 1.0}, }) + pushSlotsHeld = promauto.NewGauge(prometheus.GaugeOpts{ + Namespace: namespace, + Subsystem: "push", + Name: "slots_held", + Help: "Pushes holding a concurrency slot, so unpacking a packfile right now.", + }) + + pushQueueWaiting = promauto.NewGauge(prometheus.GaugeOpts{ + Namespace: namespace, + Subsystem: "push", + Name: "queue_waiting", + Help: "Pushes waiting for a concurrency slot. A value that stays above zero means -max-concurrent-pushes is below the offered load.", + }) + + pushQueueWait = promauto.NewHistogram(prometheus.HistogramOpts{ + Namespace: namespace, + Subsystem: "push", + Name: "queue_wait_seconds", + Help: "Time a push spent waiting for a concurrency slot.", + Buckets: []float64{0.001, 0.01, 0.1, 1, 5, 15, 30, 60, 120, 300}, + }) + + pushQueueOutcomes = promauto.NewCounterVec(prometheus.CounterOpts{ + Namespace: namespace, + Subsystem: "push", + Name: "queue_outcomes_total", + Help: "Pushes that asked for a concurrency slot, by outcome (admitted, timeout, canceled).", + }, []string{"outcome"}) + refCASRetries = promauto.NewCounter(prometheus.CounterOpts{ Namespace: namespace, Subsystem: "ref", @@ -141,6 +170,36 @@ func TrackInFlight(protocol string) func() { return func() { gitInFlight.WithLabelValues(protocol).Dec() } } +// Push-slot outcomes, as passed to ObservePushWait. +const ( + PushAdmitted = "admitted" + PushTimeout = "timeout" + PushCanceled = "canceled" +) + +// TrackPushWait increments the gauge of pushes waiting for a concurrency slot +// and returns a closure that decrements it; call the result with defer. +func TrackPushWait() func() { + pushQueueWaiting.Inc() + return func() { pushQueueWaiting.Dec() } +} + +// TrackPushSlot increments the gauge of pushes holding a concurrency slot and +// returns a closure that decrements it; call the result with defer. +func TrackPushSlot() func() { + pushSlotsHeld.Inc() + return func() { pushSlotsHeld.Dec() } +} + +// ObservePushWait records one wait for a concurrency slot. outcome is +// PushAdmitted, PushTimeout, or PushCanceled, which keeps a push that gave up +// waiting distinct from every other kind of push failure. start is when the +// wait began. +func ObservePushWait(outcome string, start time.Time) { + pushQueueOutcomes.WithLabelValues(outcome).Inc() + pushQueueWait.Observe(time.Since(start).Seconds()) +} + // ObserveGitOp records a completed git operation: status is "ok", "error", or // "denied". start is when the handler began serving it. func ObserveGitOp(protocol, service, status string, start time.Time) {