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) {