fix(githttp): cap concurrent git backends globally and per client, bound body reads
- non-blocking upload-pack and receive-pack slots answer 503 with Retry-After - per-client cap keyed on clientlimit.Key answers 429 with Retry-After - fetch bodies get a 60s read deadline, pushes a 2min idle deadline - http-backend runs pack-objects with pack.threads=1 and pack.windowMemory=256m
- commit
- a2fc532796820d58a0dff2b3656265e56ab71115
- author
- brice · · 2026-10-11 18:32 UTC
- parent
- 49d6992
- signature
- GPG Verified
CD88F9E564E0102646FB6077A1A07C7C7C3E349D· arch-gpg Verified for the committer email
4 files · +432 -5
M internal/adapters/gitcli/backend.go +7 -0
| @@ -32,6 +32,8 @@ func HooksDir(dataDir string) string { return filepath.Join(dataDir, "run", "hoo | |||
| 32 | 32 | // core.hooksPath=<data>/run/hooks | |
| 33 | 33 | // receive.fsckObjects=true | |
| 34 | 34 | // uploadpack.allowFilter=true | |
| 35 | + | // pack.threads=1 | |
| 36 | + | // pack.windowMemory=256m | |
| 35 | 37 | // http.receivepack=true (push only) | |
| 36 | 38 | // receive.maxInputSize=<max> (push only, when configured) | |
| 37 | 39 | // | |
| @@ -45,6 +47,11 @@ func (g *Git) HTTPBackend(ctx context.Context, push bool) (ports.HTTPBackend, er | |||
| 45 | 47 | {"core.hooksPath", HooksDir(g.dataDir)}, | |
| 46 | 48 | {"receive.fsckObjects", "true"}, | |
| 47 | 49 | {"uploadpack.allowFilter", "true"}, | |
| 50 | + | // pack-objects defaults to one thread per CPU and an unbounded delta | |
| 51 | + | // window; with several backends at once that is the memory and CPU | |
| 52 | + | // of the host. The backends are bounded by the handler instead. | |
| 53 | + | {"pack.threads", "1"}, | |
| 54 | + | {"pack.windowMemory", "256m"}, | |
| 48 | 55 | } | |
| 49 | 56 | if push { | |
| 50 | 57 | conf = append(conf, [2]string{"http.receivepack", "true"}) | |
| @@ -32,6 +32,8 @@ func HooksDir(dataDir string) string { return filepath.Join(dataDir, "run", "hoo | |||||
| 32 | // core.hooksPath=<data>/run/hooks | 32 | // core.hooksPath=<data>/run/hooks | ||
| 33 | // receive.fsckObjects=true | 33 | // receive.fsckObjects=true | ||
| 34 | // uploadpack.allowFilter=true | 34 | // uploadpack.allowFilter=true | ||
| 35 | + | // pack.threads=1 | |||
| 36 | + | // pack.windowMemory=256m | |||
| 35 | // http.receivepack=true (push only) | 37 | // http.receivepack=true (push only) | ||
| 36 | // receive.maxInputSize=<max> (push only, when configured) | 38 | // receive.maxInputSize=<max> (push only, when configured) | ||
| 37 | // | 39 | // | ||
| @@ -45,6 +47,11 @@ func (g *Git) HTTPBackend(ctx context.Context, push bool) (ports.HTTPBackend, er | |||||
| 45 | {"core.hooksPath", HooksDir(g.dataDir)}, | 47 | {"core.hooksPath", HooksDir(g.dataDir)}, | ||
| 46 | {"receive.fsckObjects", "true"}, | 48 | {"receive.fsckObjects", "true"}, | ||
| 47 | {"uploadpack.allowFilter", "true"}, | 49 | {"uploadpack.allowFilter", "true"}, | ||
| 50 | + | // pack-objects defaults to one thread per CPU and an unbounded delta | |||
| 51 | + | // window; with several backends at once that is the memory and CPU | |||
| 52 | + | // of the host. The backends are bounded by the handler instead. | |||
| 53 | + | {"pack.threads", "1"}, | |||
| 54 | + | {"pack.windowMemory", "256m"}, | |||
| 48 | } | 55 | } | ||
| 49 | if push { | 56 | if push { | ||
| 50 | conf = append(conf, [2]string{"http.receivepack", "true"}) | 57 | conf = append(conf, [2]string{"http.receivepack", "true"}) | ||
M internal/adapters/gitcli/backend_test.go +4 -0
| @@ -86,6 +86,8 @@ func TestHTTPBackendEnvironment(t *testing.T) { | |||
| 86 | 86 | "core.hooksPath": filepath.Join(data, "run", "hooks"), | |
| 87 | 87 | "receive.fsckObjects": "true", | |
| 88 | 88 | "uploadpack.allowFilter": "true", | |
| 89 | + | "pack.threads": "1", | |
| 90 | + | "pack.windowMemory": "256m", | |
| 89 | 91 | }, backendConf(e)) | |
| 90 | 92 | ||
| 91 | 93 | push, err := g.HTTPBackend(ctx, true) | |
| @@ -96,6 +98,8 @@ func TestHTTPBackendEnvironment(t *testing.T) { | |||
| 96 | 98 | "core.hooksPath": filepath.Join(data, "run", "hooks"), | |
| 97 | 99 | "receive.fsckObjects": "true", | |
| 98 | 100 | "uploadpack.allowFilter": "true", | |
| 101 | + | "pack.threads": "1", | |
| 102 | + | "pack.windowMemory": "256m", | |
| 99 | 103 | "http.receivepack": "true", | |
| 100 | 104 | }, backendConf(pe), "http.receivepack only for a push; no size limit unless configured") | |
| 101 | 105 | ||
| @@ -86,6 +86,8 @@ func TestHTTPBackendEnvironment(t *testing.T) { | |||||
| 86 | "core.hooksPath": filepath.Join(data, "run", "hooks"), | 86 | "core.hooksPath": filepath.Join(data, "run", "hooks"), | ||
| 87 | "receive.fsckObjects": "true", | 87 | "receive.fsckObjects": "true", | ||
| 88 | "uploadpack.allowFilter": "true", | 88 | "uploadpack.allowFilter": "true", | ||
| 89 | + | "pack.threads": "1", | |||
| 90 | + | "pack.windowMemory": "256m", | |||
| 89 | }, backendConf(e)) | 91 | }, backendConf(e)) | ||
| 90 | 92 | ||||
| 91 | push, err := g.HTTPBackend(ctx, true) | 93 | push, err := g.HTTPBackend(ctx, true) | ||
| @@ -96,6 +98,8 @@ func TestHTTPBackendEnvironment(t *testing.T) { | |||||
| 96 | "core.hooksPath": filepath.Join(data, "run", "hooks"), | 98 | "core.hooksPath": filepath.Join(data, "run", "hooks"), | ||
| 97 | "receive.fsckObjects": "true", | 99 | "receive.fsckObjects": "true", | ||
| 98 | "uploadpack.allowFilter": "true", | 100 | "uploadpack.allowFilter": "true", | ||
| 101 | + | "pack.threads": "1", | |||
| 102 | + | "pack.windowMemory": "256m", | |||
| 99 | "http.receivepack": "true", | 103 | "http.receivepack": "true", | ||
| 100 | }, backendConf(pe), "http.receivepack only for a push; no size limit unless configured") | 104 | }, backendConf(pe), "http.receivepack only for a push; no size limit unless configured") | ||
| 101 | 105 | ||||
M internal/adapters/githttp/handler.go +140 -5
| @@ -7,16 +7,21 @@ | |||
| 7 | 7 | package githttp | |
| 8 | 8 | ||
| 9 | 9 | import ( | |
| 10 | + | "cmp" | |
| 10 | 11 | "context" | |
| 11 | 12 | "errors" | |
| 13 | + | "io" | |
| 12 | 14 | "log" | |
| 13 | 15 | "math" | |
| 14 | 16 | "net/http" | |
| 15 | 17 | "net/http/cgi" | |
| 16 | 18 | "net/netip" | |
| 17 | 19 | "regexp" | |
| 20 | + | "runtime" | |
| 18 | 21 | "strconv" | |
| 19 | 22 | "strings" | |
| 23 | + | "sync" | |
| 24 | + | "time" | |
| 20 | 25 | ||
| 21 | 26 | "github.com/bnema/zerowrap" | |
| 22 | 27 | ||
| @@ -102,13 +107,48 @@ type Config struct { | |||
| 102 | 107 | // MaxBodyBytes bounds the body of a push; 0 means DefaultMaxBodyBytes. | |
| 103 | 108 | // The body of a fetch is bounded by 64 MiB at most. | |
| 104 | 109 | MaxBodyBytes int64 | |
| 110 | + | // MaxUploadPacks bounds the git-upload-pack backends (fetches, clones, | |
| 111 | + | // and the ref advertisement that precedes them) running at once; 0 means | |
| 112 | + | // 2*GOMAXPROCS. | |
| 113 | + | MaxUploadPacks int | |
| 114 | + | // MaxReceivePacks bounds the git-receive-pack backends (pushes) running | |
| 115 | + | // at once; 0 means max(2, GOMAXPROCS/2). | |
| 116 | + | MaxReceivePacks int | |
| 117 | + | // MaxPerClient bounds the backends running at once for one client (see | |
| 118 | + | // clientlimit.Key); 0 means DefaultMaxPerClient. | |
| 119 | + | MaxPerClient int | |
| 105 | 120 | } | |
| 106 | 121 | ||
| 122 | + | // Bounds on the git http-backend processes and on how long a client may take | |
| 123 | + | // to send a request body. | |
| 124 | + | const ( | |
| 125 | + | // DefaultMaxPerClient is the backends one client may run at once. | |
| 126 | + | DefaultMaxPerClient = 4 | |
| 127 | + | ||
| 128 | + | // retryAfter is the Retry-After, in seconds, of a request refused because | |
| 129 | + | // the backends are busy. | |
| 130 | + | retryAfter = "5" | |
| 131 | + | ||
| 132 | + | // fetchBodyTimeout is the whole time a client has to send the body of a | |
| 133 | + | // fetch, which is small (64 MiB at most). | |
| 134 | + | fetchBodyTimeout = 60 * time.Second | |
| 135 | + | // pushIdleTimeout is how long a push may go without sending a byte. A | |
| 136 | + | // push is bounded by its size, not by its duration, so the deadline moves | |
| 137 | + | // forward with every read. | |
| 138 | + | pushIdleTimeout = 2 * time.Minute | |
| 139 | + | ) | |
| 140 | + | ||
| 107 | 141 | // Handler serves the smart HTTP routes of Patterns. | |
| 108 | 142 | type Handler struct { | |
| 109 | 143 | cfg Config | |
| 110 | 144 | limiter *clientlimit.Limiter | |
| 111 | 145 | maxBody int64 | |
| 146 | + | ||
| 147 | + | upload, receive chan struct{} // slots of the backends, per service | |
| 148 | + | ||
| 149 | + | mu sync.Mutex // guards clients | |
| 150 | + | clients map[string]int // backends running, per clientlimit.Key | |
| 151 | + | perCli int | |
| 112 | 152 | } | |
| 113 | 153 | ||
| 114 | 154 | // New builds a Handler. | |
| @@ -120,6 +160,11 @@ func New(cfg Config) *Handler { | |||
| 120 | 160 | if h.maxBody <= 0 { | |
| 121 | 161 | h.maxBody = DefaultMaxBodyBytes | |
| 122 | 162 | } | |
| 163 | + | procs := runtime.GOMAXPROCS(0) | |
| 164 | + | h.upload = make(chan struct{}, cmp.Or(cfg.MaxUploadPacks, 2*procs)) | |
| 165 | + | h.receive = make(chan struct{}, cmp.Or(cfg.MaxReceivePacks, max(2, procs/2))) | |
| 166 | + | h.perCli = cmp.Or(cfg.MaxPerClient, DefaultMaxPerClient) | |
| 167 | + | h.clients = make(map[string]int) | |
| 123 | 168 | return h | |
| 124 | 169 | } | |
| 125 | 170 | ||
| @@ -225,7 +270,7 @@ func (h *Handler) ServeHTTP(w http.ResponseWriter, r *http.Request) { | |||
| 225 | 270 | return | |
| 226 | 271 | } | |
| 227 | 272 | ||
| 228 | - | h.serveBackend(w, r, rt, name, principal) | |
| 273 | + | h.serveBackend(w, r, rt, name, principal, clientlimit.Key(addr)) | |
| 229 | 274 | } | |
| 230 | 275 | ||
| 231 | 276 | // authenticate reads the credentials of the request. No credentials is the | |
| @@ -311,11 +356,64 @@ var inheritedEnv = []string{ | |||
| 311 | 356 | "LD_LIBRARY_PATH_32=", "LD_LIBRARY_PATH_64=", "DYLD_LIBRARY_PATH=", "SHLIB_PATH=", | |
| 312 | 357 | } | |
| 313 | 358 | ||
| 359 | + | // acquire takes a backend slot for the client and the service of rt, or | |
| 360 | + | // answers 429 (the client already runs its share) or 503 (the server runs all | |
| 361 | + | // it can) with a Retry-After and reports false. | |
| 362 | + | // | |
| 363 | + | // It never waits. A backend can run for minutes, so a queue would only trade | |
| 364 | + | // a refusal for a held connection and goroutine, and let a burst of requests | |
| 365 | + | // pile up behind slow clones. Git does not retry by itself, but the refusal is | |
| 366 | + | // cheap and says when to come back. release must be called when ok. | |
| 367 | + | func (h *Handler) acquire(ctx context.Context, w http.ResponseWriter, rt route, client string) (release func(), ok bool) { | |
| 368 | + | logger := zerowrap.FromCtx(ctx) | |
| 369 | + | refuse := func(status int, msg string) { | |
| 370 | + | w.Header().Set("Retry-After", retryAfter) | |
| 371 | + | http.Error(w, msg, status) | |
| 372 | + | } | |
| 373 | + | ||
| 374 | + | h.mu.Lock() | |
| 375 | + | if h.clients[client] >= h.perCli { | |
| 376 | + | h.mu.Unlock() | |
| 377 | + | logger.Warn().Str("client", client).Msg("too many git requests in progress for the client") | |
| 378 | + | refuse(http.StatusTooManyRequests, "too many requests in progress") | |
| 379 | + | return nil, false | |
| 380 | + | } | |
| 381 | + | h.clients[client]++ | |
| 382 | + | h.mu.Unlock() | |
| 383 | + | releaseClient := func() { | |
| 384 | + | h.mu.Lock() | |
| 385 | + | if h.clients[client]--; h.clients[client] <= 0 { | |
| 386 | + | delete(h.clients, client) | |
| 387 | + | } | |
| 388 | + | h.mu.Unlock() | |
| 389 | + | } | |
| 390 | + | ||
| 391 | + | slots := h.upload | |
| 392 | + | if rt.push() { | |
| 393 | + | slots = h.receive | |
| 394 | + | } | |
| 395 | + | select { | |
| 396 | + | case slots <- struct{}{}: | |
| 397 | + | return func() { <-slots; releaseClient() }, true | |
| 398 | + | default: | |
| 399 | + | releaseClient() | |
| 400 | + | logger.Warn().Str("service", rt.service).Msg("all git backends busy, request refused") | |
| 401 | + | refuse(http.StatusServiceUnavailable, "server busy") | |
| 402 | + | return nil, false | |
| 403 | + | } | |
| 404 | + | } | |
| 405 | + | ||
| 314 | 406 | // serveBackend runs git http-backend for an authorized request. | |
| 315 | - | func (h *Handler) serveBackend(w http.ResponseWriter, r *http.Request, rt route, name domain.RepoName, p domain.Principal) { | |
| 407 | + | func (h *Handler) serveBackend(w http.ResponseWriter, r *http.Request, rt route, name domain.RepoName, p domain.Principal, client string) { | |
| 316 | 408 | ctx := r.Context() | |
| 317 | 409 | logger := zerowrap.FromCtx(ctx) | |
| 318 | 410 | ||
| 411 | + | release, ok := h.acquire(ctx, w, rt, client) | |
| 412 | + | if !ok { | |
| 413 | + | return | |
| 414 | + | } | |
| 415 | + | defer release() | |
| 416 | + | ||
| 319 | 417 | be, err := h.cfg.Backend.HTTPBackend(ctx, rt.push()) | |
| 320 | 418 | if err != nil { | |
| 321 | 419 | logger.Error().Err(err).Msg("prepare git http-backend") | |
| @@ -324,8 +422,8 @@ func (h *Handler) serveBackend(w http.ResponseWriter, r *http.Request, rt route, | |||
| 324 | 422 | } | |
| 325 | 423 | env := append(append([]string(nil), be.Env...), inheritedEnv...) | |
| 326 | 424 | if rt.suffix == serviceReceive { | |
| 327 | - | token, release := h.cfg.Hooks.Register(ctx, name, p) | |
| 328 | - | defer release() | |
| 425 | + | token, revoke := h.cfg.Hooks.Register(ctx, name, p) | |
| 426 | + | defer revoke() | |
| 329 | 427 | env = append(env, EnvHookSocket+"="+h.cfg.Hooks.Socket(), EnvHookToken+"="+token) | |
| 330 | 428 | } | |
| 331 | 429 | ||
| @@ -383,7 +481,16 @@ func (h *Handler) backendRequest(w http.ResponseWriter, r *http.Request, rt rout | |||
| 383 | 481 | req.Header.Set("Git-Protocol", v) | |
| 384 | 482 | } | |
| 385 | 483 | ||
| 386 | - | req.Body = http.MaxBytesReader(w, r.Body, limit) | |
| 484 | + | body := &deadlineBody{ | |
| 485 | + | ReadCloser: http.MaxBytesReader(w, r.Body, limit), | |
| 486 | + | rc: http.NewResponseController(w), | |
| 487 | + | } | |
| 488 | + | if rt.push() { | |
| 489 | + | body.idle = pushIdleTimeout | |
| 490 | + | } else { | |
| 491 | + | body.until = time.Now().Add(fetchBodyTimeout) | |
| 492 | + | } | |
| 493 | + | req.Body = body | |
| 387 | 494 | if r.ContentLength < 0 || len(r.TransferEncoding) > 0 { | |
| 388 | 495 | req.ContentLength = -1 | |
| 389 | 496 | req.TransferEncoding = nil | |
| @@ -391,6 +498,34 @@ func (h *Handler) backendRequest(w http.ResponseWriter, r *http.Request, rt rout | |||
| 391 | 498 | return req, 0 | |
| 392 | 499 | } | |
| 393 | 500 | ||
| 501 | + | // deadlineBody bounds the time a client takes to send a request body, which | |
| 502 | + | // the server's own timeouts leave unbounded: a client that stalls would hold a | |
| 503 | + | // backend slot and its process forever. Until the body ends, the connection | |
| 504 | + | // read deadline is until, or when that is zero, idle after the last read. At | |
| 505 | + | // the end of the body the deadline is lifted again, so that it does not cut | |
| 506 | + | // the connection while the response is still being written. | |
| 507 | + | type deadlineBody struct { | |
| 508 | + | io.ReadCloser | |
| 509 | + | rc *http.ResponseController | |
| 510 | + | idle time.Duration // moves the deadline forward at each read, if non-zero | |
| 511 | + | until time.Time // fixed deadline, when idle is zero | |
| 512 | + | } | |
| 513 | + | ||
| 514 | + | func (b *deadlineBody) Read(p []byte) (int, error) { | |
| 515 | + | d := b.until | |
| 516 | + | if b.idle > 0 { | |
| 517 | + | d = time.Now().Add(b.idle) | |
| 518 | + | } | |
| 519 | + | // Not every ResponseWriter can set a deadline (HTTP/2 can, wrappers | |
| 520 | + | // might not): then the body is simply not bounded here. | |
| 521 | + | _ = b.rc.SetReadDeadline(d) | |
| 522 | + | n, err := b.ReadCloser.Read(p) | |
| 523 | + | if err != nil { | |
| 524 | + | _ = b.rc.SetReadDeadline(time.Time{}) | |
| 525 | + | } | |
| 526 | + | return n, err | |
| 527 | + | } | |
| 528 | + | ||
| 394 | 529 | // logWriter sends what the Git process (or the CGI layer) writes to its | |
| 395 | 530 | // standard error to the logger of the request. | |
| 396 | 531 | type logWriter struct { | |
| @@ -7,16 +7,21 @@ | |||||
| 7 | package githttp | 7 | package githttp | ||
| 8 | 8 | ||||
| 9 | import ( | 9 | import ( | ||
| 10 | + | "cmp" | |||
| 10 | "context" | 11 | "context" | ||
| 11 | "errors" | 12 | "errors" | ||
| 13 | + | "io" | |||
| 12 | "log" | 14 | "log" | ||
| 13 | "math" | 15 | "math" | ||
| 14 | "net/http" | 16 | "net/http" | ||
| 15 | "net/http/cgi" | 17 | "net/http/cgi" | ||
| 16 | "net/netip" | 18 | "net/netip" | ||
| 17 | "regexp" | 19 | "regexp" | ||
| 20 | + | "runtime" | |||
| 18 | "strconv" | 21 | "strconv" | ||
| 19 | "strings" | 22 | "strings" | ||
| 23 | + | "sync" | |||
| 24 | + | "time" | |||
| 20 | 25 | ||||
| 21 | "github.com/bnema/zerowrap" | 26 | "github.com/bnema/zerowrap" | ||
| 22 | 27 | ||||
| @@ -102,13 +107,48 @@ type Config struct { | |||||
| 102 | // MaxBodyBytes bounds the body of a push; 0 means DefaultMaxBodyBytes. | 107 | // MaxBodyBytes bounds the body of a push; 0 means DefaultMaxBodyBytes. | ||
| 103 | // The body of a fetch is bounded by 64 MiB at most. | 108 | // The body of a fetch is bounded by 64 MiB at most. | ||
| 104 | MaxBodyBytes int64 | 109 | MaxBodyBytes int64 | ||
| 110 | + | // MaxUploadPacks bounds the git-upload-pack backends (fetches, clones, | |||
| 111 | + | // and the ref advertisement that precedes them) running at once; 0 means | |||
| 112 | + | // 2*GOMAXPROCS. | |||
| 113 | + | MaxUploadPacks int | |||
| 114 | + | // MaxReceivePacks bounds the git-receive-pack backends (pushes) running | |||
| 115 | + | // at once; 0 means max(2, GOMAXPROCS/2). | |||
| 116 | + | MaxReceivePacks int | |||
| 117 | + | // MaxPerClient bounds the backends running at once for one client (see | |||
| 118 | + | // clientlimit.Key); 0 means DefaultMaxPerClient. | |||
| 119 | + | MaxPerClient int | |||
| 105 | } | 120 | } | ||
| 106 | 121 | ||||
| 122 | + | // Bounds on the git http-backend processes and on how long a client may take | |||
| 123 | + | // to send a request body. | |||
| 124 | + | const ( | |||
| 125 | + | // DefaultMaxPerClient is the backends one client may run at once. | |||
| 126 | + | DefaultMaxPerClient = 4 | |||
| 127 | + | ||||
| 128 | + | // retryAfter is the Retry-After, in seconds, of a request refused because | |||
| 129 | + | // the backends are busy. | |||
| 130 | + | retryAfter = "5" | |||
| 131 | + | ||||
| 132 | + | // fetchBodyTimeout is the whole time a client has to send the body of a | |||
| 133 | + | // fetch, which is small (64 MiB at most). | |||
| 134 | + | fetchBodyTimeout = 60 * time.Second | |||
| 135 | + | // pushIdleTimeout is how long a push may go without sending a byte. A | |||
| 136 | + | // push is bounded by its size, not by its duration, so the deadline moves | |||
| 137 | + | // forward with every read. | |||
| 138 | + | pushIdleTimeout = 2 * time.Minute | |||
| 139 | + | ) | |||
| 140 | + | ||||
| 107 | // Handler serves the smart HTTP routes of Patterns. | 141 | // Handler serves the smart HTTP routes of Patterns. | ||
| 108 | type Handler struct { | 142 | type Handler struct { | ||
| 109 | cfg Config | 143 | cfg Config | ||
| 110 | limiter *clientlimit.Limiter | 144 | limiter *clientlimit.Limiter | ||
| 111 | maxBody int64 | 145 | maxBody int64 | ||
| 146 | + | ||||
| 147 | + | upload, receive chan struct{} // slots of the backends, per service | |||
| 148 | + | ||||
| 149 | + | mu sync.Mutex // guards clients | |||
| 150 | + | clients map[string]int // backends running, per clientlimit.Key | |||
| 151 | + | perCli int | |||
| 112 | } | 152 | } | ||
| 113 | 153 | ||||
| 114 | // New builds a Handler. | 154 | // New builds a Handler. | ||
| @@ -120,6 +160,11 @@ func New(cfg Config) *Handler { | |||||
| 120 | if h.maxBody <= 0 { | 160 | if h.maxBody <= 0 { | ||
| 121 | h.maxBody = DefaultMaxBodyBytes | 161 | h.maxBody = DefaultMaxBodyBytes | ||
| 122 | } | 162 | } | ||
| 163 | + | procs := runtime.GOMAXPROCS(0) | |||
| 164 | + | h.upload = make(chan struct{}, cmp.Or(cfg.MaxUploadPacks, 2*procs)) | |||
| 165 | + | h.receive = make(chan struct{}, cmp.Or(cfg.MaxReceivePacks, max(2, procs/2))) | |||
| 166 | + | h.perCli = cmp.Or(cfg.MaxPerClient, DefaultMaxPerClient) | |||
| 167 | + | h.clients = make(map[string]int) | |||
| 123 | return h | 168 | return h | ||
| 124 | } | 169 | } | ||
| 125 | 170 | ||||
| @@ -225,7 +270,7 @@ func (h *Handler) ServeHTTP(w http.ResponseWriter, r *http.Request) { | |||||
| 225 | return | 270 | return | ||
| 226 | } | 271 | } | ||
| 227 | 272 | ||||
| 228 | - | h.serveBackend(w, r, rt, name, principal) | 273 | + | h.serveBackend(w, r, rt, name, principal, clientlimit.Key(addr)) |
| 229 | } | 274 | } | ||
| 230 | 275 | ||||
| 231 | // authenticate reads the credentials of the request. No credentials is the | 276 | // authenticate reads the credentials of the request. No credentials is the | ||
| @@ -311,11 +356,64 @@ var inheritedEnv = []string{ | |||||
| 311 | "LD_LIBRARY_PATH_32=", "LD_LIBRARY_PATH_64=", "DYLD_LIBRARY_PATH=", "SHLIB_PATH=", | 356 | "LD_LIBRARY_PATH_32=", "LD_LIBRARY_PATH_64=", "DYLD_LIBRARY_PATH=", "SHLIB_PATH=", | ||
| 312 | } | 357 | } | ||
| 313 | 358 | ||||
| 359 | + | // acquire takes a backend slot for the client and the service of rt, or | |||
| 360 | + | // answers 429 (the client already runs its share) or 503 (the server runs all | |||
| 361 | + | // it can) with a Retry-After and reports false. | |||
| 362 | + | // | |||
| 363 | + | // It never waits. A backend can run for minutes, so a queue would only trade | |||
| 364 | + | // a refusal for a held connection and goroutine, and let a burst of requests | |||
| 365 | + | // pile up behind slow clones. Git does not retry by itself, but the refusal is | |||
| 366 | + | // cheap and says when to come back. release must be called when ok. | |||
| 367 | + | func (h *Handler) acquire(ctx context.Context, w http.ResponseWriter, rt route, client string) (release func(), ok bool) { | |||
| 368 | + | logger := zerowrap.FromCtx(ctx) | |||
| 369 | + | refuse := func(status int, msg string) { | |||
| 370 | + | w.Header().Set("Retry-After", retryAfter) | |||
| 371 | + | http.Error(w, msg, status) | |||
| 372 | + | } | |||
| 373 | + | ||||
| 374 | + | h.mu.Lock() | |||
| 375 | + | if h.clients[client] >= h.perCli { | |||
| 376 | + | h.mu.Unlock() | |||
| 377 | + | logger.Warn().Str("client", client).Msg("too many git requests in progress for the client") | |||
| 378 | + | refuse(http.StatusTooManyRequests, "too many requests in progress") | |||
| 379 | + | return nil, false | |||
| 380 | + | } | |||
| 381 | + | h.clients[client]++ | |||
| 382 | + | h.mu.Unlock() | |||
| 383 | + | releaseClient := func() { | |||
| 384 | + | h.mu.Lock() | |||
| 385 | + | if h.clients[client]--; h.clients[client] <= 0 { | |||
| 386 | + | delete(h.clients, client) | |||
| 387 | + | } | |||
| 388 | + | h.mu.Unlock() | |||
| 389 | + | } | |||
| 390 | + | ||||
| 391 | + | slots := h.upload | |||
| 392 | + | if rt.push() { | |||
| 393 | + | slots = h.receive | |||
| 394 | + | } | |||
| 395 | + | select { | |||
| 396 | + | case slots <- struct{}{}: | |||
| 397 | + | return func() { <-slots; releaseClient() }, true | |||
| 398 | + | default: | |||
| 399 | + | releaseClient() | |||
| 400 | + | logger.Warn().Str("service", rt.service).Msg("all git backends busy, request refused") | |||
| 401 | + | refuse(http.StatusServiceUnavailable, "server busy") | |||
| 402 | + | return nil, false | |||
| 403 | + | } | |||
| 404 | + | } | |||
| 405 | + | ||||
| 314 | // serveBackend runs git http-backend for an authorized request. | 406 | // serveBackend runs git http-backend for an authorized request. | ||
| 315 | - | func (h *Handler) serveBackend(w http.ResponseWriter, r *http.Request, rt route, name domain.RepoName, p domain.Principal) { | 407 | + | func (h *Handler) serveBackend(w http.ResponseWriter, r *http.Request, rt route, name domain.RepoName, p domain.Principal, client string) { |
| 316 | ctx := r.Context() | 408 | ctx := r.Context() | ||
| 317 | logger := zerowrap.FromCtx(ctx) | 409 | logger := zerowrap.FromCtx(ctx) | ||
| 318 | 410 | ||||
| 411 | + | release, ok := h.acquire(ctx, w, rt, client) | |||
| 412 | + | if !ok { | |||
| 413 | + | return | |||
| 414 | + | } | |||
| 415 | + | defer release() | |||
| 416 | + | ||||
| 319 | be, err := h.cfg.Backend.HTTPBackend(ctx, rt.push()) | 417 | be, err := h.cfg.Backend.HTTPBackend(ctx, rt.push()) | ||
| 320 | if err != nil { | 418 | if err != nil { | ||
| 321 | logger.Error().Err(err).Msg("prepare git http-backend") | 419 | logger.Error().Err(err).Msg("prepare git http-backend") | ||
| @@ -324,8 +422,8 @@ func (h *Handler) serveBackend(w http.ResponseWriter, r *http.Request, rt route, | |||||
| 324 | } | 422 | } | ||
| 325 | env := append(append([]string(nil), be.Env...), inheritedEnv...) | 423 | env := append(append([]string(nil), be.Env...), inheritedEnv...) | ||
| 326 | if rt.suffix == serviceReceive { | 424 | if rt.suffix == serviceReceive { | ||
| 327 | - | token, release := h.cfg.Hooks.Register(ctx, name, p) | 425 | + | token, revoke := h.cfg.Hooks.Register(ctx, name, p) |
| 328 | - | defer release() | 426 | + | defer revoke() |
| 329 | env = append(env, EnvHookSocket+"="+h.cfg.Hooks.Socket(), EnvHookToken+"="+token) | 427 | env = append(env, EnvHookSocket+"="+h.cfg.Hooks.Socket(), EnvHookToken+"="+token) | ||
| 330 | } | 428 | } | ||
| 331 | 429 | ||||
| @@ -383,7 +481,16 @@ func (h *Handler) backendRequest(w http.ResponseWriter, r *http.Request, rt rout | |||||
| 383 | req.Header.Set("Git-Protocol", v) | 481 | req.Header.Set("Git-Protocol", v) | ||
| 384 | } | 482 | } | ||
| 385 | 483 | ||||
| 386 | - | req.Body = http.MaxBytesReader(w, r.Body, limit) | 484 | + | body := &deadlineBody{ |
| 485 | + | ReadCloser: http.MaxBytesReader(w, r.Body, limit), | |||
| 486 | + | rc: http.NewResponseController(w), | |||
| 487 | + | } | |||
| 488 | + | if rt.push() { | |||
| 489 | + | body.idle = pushIdleTimeout | |||
| 490 | + | } else { | |||
| 491 | + | body.until = time.Now().Add(fetchBodyTimeout) | |||
| 492 | + | } | |||
| 493 | + | req.Body = body | |||
| 387 | if r.ContentLength < 0 || len(r.TransferEncoding) > 0 { | 494 | if r.ContentLength < 0 || len(r.TransferEncoding) > 0 { | ||
| 388 | req.ContentLength = -1 | 495 | req.ContentLength = -1 | ||
| 389 | req.TransferEncoding = nil | 496 | req.TransferEncoding = nil | ||
| @@ -391,6 +498,34 @@ func (h *Handler) backendRequest(w http.ResponseWriter, r *http.Request, rt rout | |||||
| 391 | return req, 0 | 498 | return req, 0 | ||
| 392 | } | 499 | } | ||
| 393 | 500 | ||||
| 501 | + | // deadlineBody bounds the time a client takes to send a request body, which | |||
| 502 | + | // the server's own timeouts leave unbounded: a client that stalls would hold a | |||
| 503 | + | // backend slot and its process forever. Until the body ends, the connection | |||
| 504 | + | // read deadline is until, or when that is zero, idle after the last read. At | |||
| 505 | + | // the end of the body the deadline is lifted again, so that it does not cut | |||
| 506 | + | // the connection while the response is still being written. | |||
| 507 | + | type deadlineBody struct { | |||
| 508 | + | io.ReadCloser | |||
| 509 | + | rc *http.ResponseController | |||
| 510 | + | idle time.Duration // moves the deadline forward at each read, if non-zero | |||
| 511 | + | until time.Time // fixed deadline, when idle is zero | |||
| 512 | + | } | |||
| 513 | + | ||||
| 514 | + | func (b *deadlineBody) Read(p []byte) (int, error) { | |||
| 515 | + | d := b.until | |||
| 516 | + | if b.idle > 0 { | |||
| 517 | + | d = time.Now().Add(b.idle) | |||
| 518 | + | } | |||
| 519 | + | // Not every ResponseWriter can set a deadline (HTTP/2 can, wrappers | |||
| 520 | + | // might not): then the body is simply not bounded here. | |||
| 521 | + | _ = b.rc.SetReadDeadline(d) | |||
| 522 | + | n, err := b.ReadCloser.Read(p) | |||
| 523 | + | if err != nil { | |||
| 524 | + | _ = b.rc.SetReadDeadline(time.Time{}) | |||
| 525 | + | } | |||
| 526 | + | return n, err | |||
| 527 | + | } | |||
| 528 | + | ||||
| 394 | // logWriter sends what the Git process (or the CGI layer) writes to its | 529 | // logWriter sends what the Git process (or the CGI layer) writes to its | ||
| 395 | // standard error to the logger of the request. | 530 | // standard error to the logger of the request. | ||
| 396 | type logWriter struct { | 531 | type logWriter struct { | ||
A internal/adapters/githttp/limits_test.go +281 -0
| @@ -0,0 +1,281 @@ | |||
| 1 | + | // SPDX-License-Identifier: AGPL-3.0-or-later | |
| 2 | + | ||
| 3 | + | package githttp_test | |
| 4 | + | ||
| 5 | + | import ( | |
| 6 | + | "context" | |
| 7 | + | "net/http" | |
| 8 | + | "net/http/httptest" | |
| 9 | + | "net/netip" | |
| 10 | + | "path/filepath" | |
| 11 | + | "strings" | |
| 12 | + | "sync" | |
| 13 | + | "testing" | |
| 14 | + | "time" | |
| 15 | + | ||
| 16 | + | "github.com/stretchr/testify/assert" | |
| 17 | + | "github.com/stretchr/testify/mock" | |
| 18 | + | "github.com/stretchr/testify/require" | |
| 19 | + | ||
| 20 | + | "bnema.dev/ermite/internal/adapters/githttp" | |
| 21 | + | "bnema.dev/ermite/internal/ports" | |
| 22 | + | "bnema.dev/ermite/internal/ports/mocks" | |
| 23 | + | ) | |
| 24 | + | ||
| 25 | + | const ( | |
| 26 | + | pathUpload = "/pub.git/git-upload-pack" | |
| 27 | + | pathRefs = "/pub.git/info/refs?service=git-upload-pack" | |
| 28 | + | pathReceive = "/pub.git/git-receive-pack" | |
| 29 | + | ) | |
| 30 | + | ||
| 31 | + | // held is a request kept inside the backend until the gate opens. | |
| 32 | + | type held struct{ path, client string } | |
| 33 | + | ||
| 34 | + | // gatedRig is a rig whose backend blocks every request that reaches it until | |
| 35 | + | // open is called, so a test decides exactly which slots are taken. | |
| 36 | + | type gatedRig struct { | |
| 37 | + | *rig | |
| 38 | + | entered chan struct{} // a request reached the backend | |
| 39 | + | served chan struct{} // a request left the handler, its slots released | |
| 40 | + | gate chan struct{} | |
| 41 | + | done sync.WaitGroup | |
| 42 | + | } | |
| 43 | + | ||
| 44 | + | func newGatedRig(t *testing.T, mut func(*githttp.Config)) *gatedRig { | |
| 45 | + | t.Helper() | |
| 46 | + | g := &gatedRig{entered: make(chan struct{}, 64), served: make(chan struct{}, 64), gate: make(chan struct{})} | |
| 47 | + | script := cgiScript(t) | |
| 48 | + | be := mocks.NewGitBackend(t) | |
| 49 | + | be.EXPECT().HTTPBackend(mock.Anything, mock.Anything).RunAndReturn( | |
| 50 | + | func(ctx context.Context, _ bool) (ports.HTTPBackend, error) { | |
| 51 | + | g.entered <- struct{}{} | |
| 52 | + | select { | |
| 53 | + | case <-g.gate: | |
| 54 | + | case <-ctx.Done(): | |
| 55 | + | } | |
| 56 | + | return ports.HTTPBackend{Path: script, Dir: filepath.Dir(script), Env: []string{"PATH=/usr/bin:/bin"}}, nil | |
| 57 | + | }).Maybe() | |
| 58 | + | g.rig = newRig(t, func(c *githttp.Config) { | |
| 59 | + | c.Backend = be | |
| 60 | + | c.TrustedProxies = []netip.Prefix{netip.MustParsePrefix("127.0.0.0/8")} | |
| 61 | + | if mut != nil { | |
| 62 | + | mut(c) | |
| 63 | + | } | |
| 64 | + | }) | |
| 65 | + | // A client sees the response before the handler returns and gives its | |
| 66 | + | // slots back: serve through a wrapper that tells when it has. | |
| 67 | + | g.srv = httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { | |
| 68 | + | defer func() { g.served <- struct{}{} }() | |
| 69 | + | g.h.ServeHTTP(w, r) | |
| 70 | + | })) | |
| 71 | + | t.Cleanup(g.srv.Close) | |
| 72 | + | t.Cleanup(func() { g.open(); g.done.Wait() }) | |
| 73 | + | return g | |
| 74 | + | } | |
| 75 | + | ||
| 76 | + | // waitServed waits until n requests have left the handler. | |
| 77 | + | func (g *gatedRig) waitServed(n int) { | |
| 78 | + | for range n { | |
| 79 | + | <-g.served | |
| 80 | + | } | |
| 81 | + | } | |
| 82 | + | ||
| 83 | + | // hold starts the requests and returns once all of them are inside the backend. | |
| 84 | + | func (g *gatedRig) hold(t *testing.T, reqs []held) { | |
| 85 | + | t.Helper() | |
| 86 | + | for _, h := range reqs { | |
| 87 | + | method := http.MethodPost | |
| 88 | + | if h.path == pathRefs { | |
| 89 | + | method = http.MethodGet | |
| 90 | + | } | |
| 91 | + | g.done.Go(func() { | |
| 92 | + | req, err := http.NewRequest(method, g.srv.URL+h.path, nil) | |
| 93 | + | if err != nil { | |
| 94 | + | return | |
| 95 | + | } | |
| 96 | + | req.SetBasicAuth("u", "root-secret") | |
| 97 | + | req.Header.Set("X-Forwarded-For", h.client) | |
| 98 | + | if resp, err := http.DefaultClient.Do(req); err == nil { | |
| 99 | + | _ = resp.Body.Close() | |
| 100 | + | } | |
| 101 | + | }) | |
| 102 | + | } | |
| 103 | + | for range reqs { | |
| 104 | + | <-g.entered | |
| 105 | + | } | |
| 106 | + | } | |
| 107 | + | ||
| 108 | + | func (g *gatedRig) open() { | |
| 109 | + | select { | |
| 110 | + | case <-g.gate: | |
| 111 | + | default: | |
| 112 | + | close(g.gate) | |
| 113 | + | } | |
| 114 | + | } | |
| 115 | + | ||
| 116 | + | func TestHandlerBackendLimits(t *testing.T) { | |
| 117 | + | tests := []struct { | |
| 118 | + | name string | |
| 119 | + | cfg func(*githttp.Config) | |
| 120 | + | hold []held | |
| 121 | + | probe held | |
| 122 | + | want int | |
| 123 | + | }{ | |
| 124 | + | { | |
| 125 | + | name: "free upload slot", | |
| 126 | + | cfg: func(c *githttp.Config) { c.MaxUploadPacks = 2 }, | |
| 127 | + | hold: []held{{pathUpload, "198.51.100.1"}}, | |
| 128 | + | probe: held{pathUpload, "198.51.100.2"}, | |
| 129 | + | want: http.StatusOK, | |
| 130 | + | }, | |
| 131 | + | { | |
| 132 | + | name: "upload slots full", | |
| 133 | + | cfg: func(c *githttp.Config) { c.MaxUploadPacks = 2 }, | |
| 134 | + | hold: []held{{pathUpload, "198.51.100.1"}, {pathRefs, "198.51.100.2"}}, | |
| 135 | + | probe: held{pathUpload, "198.51.100.3"}, | |
| 136 | + | want: http.StatusServiceUnavailable, | |
| 137 | + | }, | |
| 138 | + | { | |
| 139 | + | name: "ref advertisement counts as an upload", | |
| 140 | + | cfg: func(c *githttp.Config) { c.MaxUploadPacks = 1 }, | |
| 141 | + | hold: []held{{pathUpload, "198.51.100.1"}}, | |
| 142 | + | probe: held{pathRefs, "198.51.100.2"}, | |
| 143 | + | want: http.StatusServiceUnavailable, | |
| 144 | + | }, | |
| 145 | + | { | |
| 146 | + | name: "receive slots full", | |
| 147 | + | cfg: func(c *githttp.Config) { c.MaxReceivePacks = 1 }, | |
| 148 | + | hold: []held{{pathReceive, "198.51.100.1"}}, | |
| 149 | + | probe: held{pathReceive, "198.51.100.2"}, | |
| 150 | + | want: http.StatusServiceUnavailable, | |
| 151 | + | }, | |
| 152 | + | { | |
| 153 | + | name: "pushes do not take upload slots", | |
| 154 | + | cfg: func(c *githttp.Config) { c.MaxUploadPacks = 1; c.MaxReceivePacks = 1 }, | |
| 155 | + | hold: []held{{pathUpload, "198.51.100.1"}}, | |
| 156 | + | probe: held{pathReceive, "198.51.100.2"}, | |
| 157 | + | want: http.StatusOK, | |
| 158 | + | }, | |
| 159 | + | { | |
| 160 | + | name: "fetches do not take receive slots", | |
| 161 | + | cfg: func(c *githttp.Config) { c.MaxUploadPacks = 1; c.MaxReceivePacks = 1 }, | |
| 162 | + | hold: []held{{pathReceive, "198.51.100.1"}}, | |
| 163 | + | probe: held{pathUpload, "198.51.100.2"}, | |
| 164 | + | want: http.StatusOK, | |
| 165 | + | }, | |
| 166 | + | { | |
| 167 | + | name: "client at its cap", | |
| 168 | + | cfg: func(c *githttp.Config) { c.MaxPerClient = 2 }, | |
| 169 | + | hold: []held{{pathUpload, "198.51.100.1"}, {pathReceive, "198.51.100.1"}}, | |
| 170 | + | probe: held{pathUpload, "198.51.100.1"}, | |
| 171 | + | want: http.StatusTooManyRequests, | |
| 172 | + | }, | |
| 173 | + | { | |
| 174 | + | name: "another client is not affected by the cap", | |
| 175 | + | cfg: func(c *githttp.Config) { c.MaxPerClient = 2 }, | |
| 176 | + | hold: []held{{pathUpload, "198.51.100.1"}, {pathUpload, "198.51.100.1"}}, | |
| 177 | + | probe: held{pathUpload, "198.51.100.2"}, | |
| 178 | + | want: http.StatusOK, | |
| 179 | + | }, | |
| 180 | + | { | |
| 181 | + | name: "addresses of one IPv6 /64 are one client", | |
| 182 | + | cfg: func(c *githttp.Config) { c.MaxPerClient = 1 }, | |
| 183 | + | hold: []held{{pathUpload, "2001:db8::1"}}, | |
| 184 | + | probe: held{pathUpload, "2001:db8::2"}, | |
| 185 | + | want: http.StatusTooManyRequests, | |
| 186 | + | }, | |
| 187 | + | } | |
| 188 | + | for _, tt := range tests { | |
| 189 | + | t.Run(tt.name, func(t *testing.T) { | |
| 190 | + | g := newGatedRig(t, func(c *githttp.Config) { | |
| 191 | + | c.MaxUploadPacks, c.MaxReceivePacks, c.MaxPerClient = 8, 8, 8 | |
| 192 | + | tt.cfg(c) | |
| 193 | + | }) | |
| 194 | + | g.hold(t, tt.hold) | |
| 195 | + | if tt.want == http.StatusOK { | |
| 196 | + | g.open() // the probe has to get through the backend gate too | |
| 197 | + | } | |
| 198 | + | code, _, hdr := g.do(t, http.MethodPost, tt.probe.path, "u", "root-secret", nil, | |
| 199 | + | map[string]string{"X-Forwarded-For": tt.probe.client}) | |
| 200 | + | if tt.probe.path == pathRefs { | |
| 201 | + | code, _, hdr = g.do(t, http.MethodGet, tt.probe.path, "u", "root-secret", nil, | |
| 202 | + | map[string]string{"X-Forwarded-For": tt.probe.client}) | |
| 203 | + | } | |
| 204 | + | assert.Equal(t, tt.want, code) | |
| 205 | + | if tt.want != http.StatusOK { | |
| 206 | + | assert.NotEmpty(t, hdr.Get("Retry-After")) | |
| 207 | + | } | |
| 208 | + | }) | |
| 209 | + | } | |
| 210 | + | } | |
| 211 | + | ||
| 212 | + | func TestHandlerBackendSlotsAreReleased(t *testing.T) { | |
| 213 | + | g := newGatedRig(t, func(c *githttp.Config) { | |
| 214 | + | c.MaxUploadPacks, c.MaxReceivePacks, c.MaxPerClient = 1, 1, 1 | |
| 215 | + | }) | |
| 216 | + | hdr := map[string]string{"X-Forwarded-For": "198.51.100.1"} | |
| 217 | + | ||
| 218 | + | g.hold(t, []held{{pathUpload, "198.51.100.1"}}) | |
| 219 | + | code, _, _ := g.do(t, http.MethodPost, pathUpload, "u", "root-secret", nil, hdr) | |
| 220 | + | require.Equal(t, http.StatusTooManyRequests, code) | |
| 221 | + | g.waitServed(1) | |
| 222 | + | ||
| 223 | + | g.open() | |
| 224 | + | g.waitServed(1) | |
| 225 | + | for _, p := range []string{pathUpload, pathReceive, pathUpload} { | |
| 226 | + | code, _, _ = g.do(t, http.MethodPost, p, "u", "root-secret", nil, hdr) | |
| 227 | + | assert.Equal(t, http.StatusOK, code, "the slots came back: %s", p) | |
| 228 | + | g.waitServed(1) | |
| 229 | + | } | |
| 230 | + | } | |
| 231 | + | ||
| 232 | + | func TestHandlerBackendFailureReleasesSlots(t *testing.T) { | |
| 233 | + | r := newRig(t, func(c *githttp.Config) { c.MaxUploadPacks, c.MaxPerClient = 1, 1 }) | |
| 234 | + | r.backendErr = context.DeadlineExceeded | |
| 235 | + | for range 3 { | |
| 236 | + | code, _, _ := r.do(t, http.MethodPost, pathUpload, "", "", nil, nil) | |
| 237 | + | assert.Equal(t, http.StatusInternalServerError, code, "never a refusal: the slot is released on failure") | |
| 238 | + | } | |
| 239 | + | } | |
| 240 | + | ||
| 241 | + | // deadlineRecorder is a ResponseWriter that supports the read deadline, as | |
| 242 | + | // the writer of net/http does, and records it. | |
| 243 | + | type deadlineRecorder struct { | |
| 244 | + | *httptest.ResponseRecorder | |
| 245 | + | deadlines []time.Time | |
| 246 | + | } | |
| 247 | + | ||
| 248 | + | func (d *deadlineRecorder) SetReadDeadline(t time.Time) error { | |
| 249 | + | d.deadlines = append(d.deadlines, t) | |
| 250 | + | return nil | |
| 251 | + | } | |
| 252 | + | ||
| 253 | + | func TestHandlerBodyReadDeadline(t *testing.T) { | |
| 254 | + | tests := []struct { | |
| 255 | + | name string | |
| 256 | + | path string | |
| 257 | + | max time.Duration | |
| 258 | + | min time.Duration | |
| 259 | + | }{ | |
| 260 | + | {"fetch body", pathUpload, time.Minute, 30 * time.Second}, | |
| 261 | + | {"push body is bounded by idleness", pathReceive, 5 * time.Minute, time.Minute}, | |
| 262 | + | } | |
| 263 | + | for _, tt := range tests { | |
| 264 | + | t.Run(tt.name, func(t *testing.T) { | |
| 265 | + | r := newRig(t, nil) | |
| 266 | + | req := httptest.NewRequest(http.MethodPost, tt.path, stringsReader("payload")) | |
| 267 | + | req.SetBasicAuth("u", "root-secret") | |
| 268 | + | w := &deadlineRecorder{ResponseRecorder: httptest.NewRecorder()} | |
| 269 | + | start := time.Now() | |
| 270 | + | r.h.ServeHTTP(w, req) | |
| 271 | + | require.Equal(t, http.StatusOK, w.Code) | |
| 272 | + | ||
| 273 | + | require.GreaterOrEqual(t, len(w.deadlines), 2) | |
| 274 | + | first, last := w.deadlines[0], w.deadlines[len(w.deadlines)-1] | |
| 275 | + | assert.WithinRange(t, first, start.Add(tt.min), time.Now().Add(tt.max)) | |
| 276 | + | assert.True(t, last.IsZero(), "the deadline is lifted once the body is read") | |
| 277 | + | }) | |
| 278 | + | } | |
| 279 | + | } | |
| 280 | + | ||
| 281 | + | func stringsReader(s string) *strings.Reader { return strings.NewReader(s) } | |
| @@ -0,0 +1,281 @@ | |||||
| 1 | + | // SPDX-License-Identifier: AGPL-3.0-or-later | |||
| 2 | + | ||||
| 3 | + | package githttp_test | |||
| 4 | + | ||||
| 5 | + | import ( | |||
| 6 | + | "context" | |||
| 7 | + | "net/http" | |||
| 8 | + | "net/http/httptest" | |||
| 9 | + | "net/netip" | |||
| 10 | + | "path/filepath" | |||
| 11 | + | "strings" | |||
| 12 | + | "sync" | |||
| 13 | + | "testing" | |||
| 14 | + | "time" | |||
| 15 | + | ||||
| 16 | + | "github.com/stretchr/testify/assert" | |||
| 17 | + | "github.com/stretchr/testify/mock" | |||
| 18 | + | "github.com/stretchr/testify/require" | |||
| 19 | + | ||||
| 20 | + | "bnema.dev/ermite/internal/adapters/githttp" | |||
| 21 | + | "bnema.dev/ermite/internal/ports" | |||
| 22 | + | "bnema.dev/ermite/internal/ports/mocks" | |||
| 23 | + | ) | |||
| 24 | + | ||||
| 25 | + | const ( | |||
| 26 | + | pathUpload = "/pub.git/git-upload-pack" | |||
| 27 | + | pathRefs = "/pub.git/info/refs?service=git-upload-pack" | |||
| 28 | + | pathReceive = "/pub.git/git-receive-pack" | |||
| 29 | + | ) | |||
| 30 | + | ||||
| 31 | + | // held is a request kept inside the backend until the gate opens. | |||
| 32 | + | type held struct{ path, client string } | |||
| 33 | + | ||||
| 34 | + | // gatedRig is a rig whose backend blocks every request that reaches it until | |||
| 35 | + | // open is called, so a test decides exactly which slots are taken. | |||
| 36 | + | type gatedRig struct { | |||
| 37 | + | *rig | |||
| 38 | + | entered chan struct{} // a request reached the backend | |||
| 39 | + | served chan struct{} // a request left the handler, its slots released | |||
| 40 | + | gate chan struct{} | |||
| 41 | + | done sync.WaitGroup | |||
| 42 | + | } | |||
| 43 | + | ||||
| 44 | + | func newGatedRig(t *testing.T, mut func(*githttp.Config)) *gatedRig { | |||
| 45 | + | t.Helper() | |||
| 46 | + | g := &gatedRig{entered: make(chan struct{}, 64), served: make(chan struct{}, 64), gate: make(chan struct{})} | |||
| 47 | + | script := cgiScript(t) | |||
| 48 | + | be := mocks.NewGitBackend(t) | |||
| 49 | + | be.EXPECT().HTTPBackend(mock.Anything, mock.Anything).RunAndReturn( | |||
| 50 | + | func(ctx context.Context, _ bool) (ports.HTTPBackend, error) { | |||
| 51 | + | g.entered <- struct{}{} | |||
| 52 | + | select { | |||
| 53 | + | case <-g.gate: | |||
| 54 | + | case <-ctx.Done(): | |||
| 55 | + | } | |||
| 56 | + | return ports.HTTPBackend{Path: script, Dir: filepath.Dir(script), Env: []string{"PATH=/usr/bin:/bin"}}, nil | |||
| 57 | + | }).Maybe() | |||
| 58 | + | g.rig = newRig(t, func(c *githttp.Config) { | |||
| 59 | + | c.Backend = be | |||
| 60 | + | c.TrustedProxies = []netip.Prefix{netip.MustParsePrefix("127.0.0.0/8")} | |||
| 61 | + | if mut != nil { | |||
| 62 | + | mut(c) | |||
| 63 | + | } | |||
| 64 | + | }) | |||
| 65 | + | // A client sees the response before the handler returns and gives its | |||
| 66 | + | // slots back: serve through a wrapper that tells when it has. | |||
| 67 | + | g.srv = httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { | |||
| 68 | + | defer func() { g.served <- struct{}{} }() | |||
| 69 | + | g.h.ServeHTTP(w, r) | |||
| 70 | + | })) | |||
| 71 | + | t.Cleanup(g.srv.Close) | |||
| 72 | + | t.Cleanup(func() { g.open(); g.done.Wait() }) | |||
| 73 | + | return g | |||
| 74 | + | } | |||
| 75 | + | ||||
| 76 | + | // waitServed waits until n requests have left the handler. | |||
| 77 | + | func (g *gatedRig) waitServed(n int) { | |||
| 78 | + | for range n { | |||
| 79 | + | <-g.served | |||
| 80 | + | } | |||
| 81 | + | } | |||
| 82 | + | ||||
| 83 | + | // hold starts the requests and returns once all of them are inside the backend. | |||
| 84 | + | func (g *gatedRig) hold(t *testing.T, reqs []held) { | |||
| 85 | + | t.Helper() | |||
| 86 | + | for _, h := range reqs { | |||
| 87 | + | method := http.MethodPost | |||
| 88 | + | if h.path == pathRefs { | |||
| 89 | + | method = http.MethodGet | |||
| 90 | + | } | |||
| 91 | + | g.done.Go(func() { | |||
| 92 | + | req, err := http.NewRequest(method, g.srv.URL+h.path, nil) | |||
| 93 | + | if err != nil { | |||
| 94 | + | return | |||
| 95 | + | } | |||
| 96 | + | req.SetBasicAuth("u", "root-secret") | |||
| 97 | + | req.Header.Set("X-Forwarded-For", h.client) | |||
| 98 | + | if resp, err := http.DefaultClient.Do(req); err == nil { | |||
| 99 | + | _ = resp.Body.Close() | |||
| 100 | + | } | |||
| 101 | + | }) | |||
| 102 | + | } | |||
| 103 | + | for range reqs { | |||
| 104 | + | <-g.entered | |||
| 105 | + | } | |||
| 106 | + | } | |||
| 107 | + | ||||
| 108 | + | func (g *gatedRig) open() { | |||
| 109 | + | select { | |||
| 110 | + | case <-g.gate: | |||
| 111 | + | default: | |||
| 112 | + | close(g.gate) | |||
| 113 | + | } | |||
| 114 | + | } | |||
| 115 | + | ||||
| 116 | + | func TestHandlerBackendLimits(t *testing.T) { | |||
| 117 | + | tests := []struct { | |||
| 118 | + | name string | |||
| 119 | + | cfg func(*githttp.Config) | |||
| 120 | + | hold []held | |||
| 121 | + | probe held | |||
| 122 | + | want int | |||
| 123 | + | }{ | |||
| 124 | + | { | |||
| 125 | + | name: "free upload slot", | |||
| 126 | + | cfg: func(c *githttp.Config) { c.MaxUploadPacks = 2 }, | |||
| 127 | + | hold: []held{{pathUpload, "198.51.100.1"}}, | |||
| 128 | + | probe: held{pathUpload, "198.51.100.2"}, | |||
| 129 | + | want: http.StatusOK, | |||
| 130 | + | }, | |||
| 131 | + | { | |||
| 132 | + | name: "upload slots full", | |||
| 133 | + | cfg: func(c *githttp.Config) { c.MaxUploadPacks = 2 }, | |||
| 134 | + | hold: []held{{pathUpload, "198.51.100.1"}, {pathRefs, "198.51.100.2"}}, | |||
| 135 | + | probe: held{pathUpload, "198.51.100.3"}, | |||
| 136 | + | want: http.StatusServiceUnavailable, | |||
| 137 | + | }, | |||
| 138 | + | { | |||
| 139 | + | name: "ref advertisement counts as an upload", | |||
| 140 | + | cfg: func(c *githttp.Config) { c.MaxUploadPacks = 1 }, | |||
| 141 | + | hold: []held{{pathUpload, "198.51.100.1"}}, | |||
| 142 | + | probe: held{pathRefs, "198.51.100.2"}, | |||
| 143 | + | want: http.StatusServiceUnavailable, | |||
| 144 | + | }, | |||
| 145 | + | { | |||
| 146 | + | name: "receive slots full", | |||
| 147 | + | cfg: func(c *githttp.Config) { c.MaxReceivePacks = 1 }, | |||
| 148 | + | hold: []held{{pathReceive, "198.51.100.1"}}, | |||
| 149 | + | probe: held{pathReceive, "198.51.100.2"}, | |||
| 150 | + | want: http.StatusServiceUnavailable, | |||
| 151 | + | }, | |||
| 152 | + | { | |||
| 153 | + | name: "pushes do not take upload slots", | |||
| 154 | + | cfg: func(c *githttp.Config) { c.MaxUploadPacks = 1; c.MaxReceivePacks = 1 }, | |||
| 155 | + | hold: []held{{pathUpload, "198.51.100.1"}}, | |||
| 156 | + | probe: held{pathReceive, "198.51.100.2"}, | |||
| 157 | + | want: http.StatusOK, | |||
| 158 | + | }, | |||
| 159 | + | { | |||
| 160 | + | name: "fetches do not take receive slots", | |||
| 161 | + | cfg: func(c *githttp.Config) { c.MaxUploadPacks = 1; c.MaxReceivePacks = 1 }, | |||
| 162 | + | hold: []held{{pathReceive, "198.51.100.1"}}, | |||
| 163 | + | probe: held{pathUpload, "198.51.100.2"}, | |||
| 164 | + | want: http.StatusOK, | |||
| 165 | + | }, | |||
| 166 | + | { | |||
| 167 | + | name: "client at its cap", | |||
| 168 | + | cfg: func(c *githttp.Config) { c.MaxPerClient = 2 }, | |||
| 169 | + | hold: []held{{pathUpload, "198.51.100.1"}, {pathReceive, "198.51.100.1"}}, | |||
| 170 | + | probe: held{pathUpload, "198.51.100.1"}, | |||
| 171 | + | want: http.StatusTooManyRequests, | |||
| 172 | + | }, | |||
| 173 | + | { | |||
| 174 | + | name: "another client is not affected by the cap", | |||
| 175 | + | cfg: func(c *githttp.Config) { c.MaxPerClient = 2 }, | |||
| 176 | + | hold: []held{{pathUpload, "198.51.100.1"}, {pathUpload, "198.51.100.1"}}, | |||
| 177 | + | probe: held{pathUpload, "198.51.100.2"}, | |||
| 178 | + | want: http.StatusOK, | |||
| 179 | + | }, | |||
| 180 | + | { | |||
| 181 | + | name: "addresses of one IPv6 /64 are one client", | |||
| 182 | + | cfg: func(c *githttp.Config) { c.MaxPerClient = 1 }, | |||
| 183 | + | hold: []held{{pathUpload, "2001:db8::1"}}, | |||
| 184 | + | probe: held{pathUpload, "2001:db8::2"}, | |||
| 185 | + | want: http.StatusTooManyRequests, | |||
| 186 | + | }, | |||
| 187 | + | } | |||
| 188 | + | for _, tt := range tests { | |||
| 189 | + | t.Run(tt.name, func(t *testing.T) { | |||
| 190 | + | g := newGatedRig(t, func(c *githttp.Config) { | |||
| 191 | + | c.MaxUploadPacks, c.MaxReceivePacks, c.MaxPerClient = 8, 8, 8 | |||
| 192 | + | tt.cfg(c) | |||
| 193 | + | }) | |||
| 194 | + | g.hold(t, tt.hold) | |||
| 195 | + | if tt.want == http.StatusOK { | |||
| 196 | + | g.open() // the probe has to get through the backend gate too | |||
| 197 | + | } | |||
| 198 | + | code, _, hdr := g.do(t, http.MethodPost, tt.probe.path, "u", "root-secret", nil, | |||
| 199 | + | map[string]string{"X-Forwarded-For": tt.probe.client}) | |||
| 200 | + | if tt.probe.path == pathRefs { | |||
| 201 | + | code, _, hdr = g.do(t, http.MethodGet, tt.probe.path, "u", "root-secret", nil, | |||
| 202 | + | map[string]string{"X-Forwarded-For": tt.probe.client}) | |||
| 203 | + | } | |||
| 204 | + | assert.Equal(t, tt.want, code) | |||
| 205 | + | if tt.want != http.StatusOK { | |||
| 206 | + | assert.NotEmpty(t, hdr.Get("Retry-After")) | |||
| 207 | + | } | |||
| 208 | + | }) | |||
| 209 | + | } | |||
| 210 | + | } | |||
| 211 | + | ||||
| 212 | + | func TestHandlerBackendSlotsAreReleased(t *testing.T) { | |||
| 213 | + | g := newGatedRig(t, func(c *githttp.Config) { | |||
| 214 | + | c.MaxUploadPacks, c.MaxReceivePacks, c.MaxPerClient = 1, 1, 1 | |||
| 215 | + | }) | |||
| 216 | + | hdr := map[string]string{"X-Forwarded-For": "198.51.100.1"} | |||
| 217 | + | ||||
| 218 | + | g.hold(t, []held{{pathUpload, "198.51.100.1"}}) | |||
| 219 | + | code, _, _ := g.do(t, http.MethodPost, pathUpload, "u", "root-secret", nil, hdr) | |||
| 220 | + | require.Equal(t, http.StatusTooManyRequests, code) | |||
| 221 | + | g.waitServed(1) | |||
| 222 | + | ||||
| 223 | + | g.open() | |||
| 224 | + | g.waitServed(1) | |||
| 225 | + | for _, p := range []string{pathUpload, pathReceive, pathUpload} { | |||
| 226 | + | code, _, _ = g.do(t, http.MethodPost, p, "u", "root-secret", nil, hdr) | |||
| 227 | + | assert.Equal(t, http.StatusOK, code, "the slots came back: %s", p) | |||
| 228 | + | g.waitServed(1) | |||
| 229 | + | } | |||
| 230 | + | } | |||
| 231 | + | ||||
| 232 | + | func TestHandlerBackendFailureReleasesSlots(t *testing.T) { | |||
| 233 | + | r := newRig(t, func(c *githttp.Config) { c.MaxUploadPacks, c.MaxPerClient = 1, 1 }) | |||
| 234 | + | r.backendErr = context.DeadlineExceeded | |||
| 235 | + | for range 3 { | |||
| 236 | + | code, _, _ := r.do(t, http.MethodPost, pathUpload, "", "", nil, nil) | |||
| 237 | + | assert.Equal(t, http.StatusInternalServerError, code, "never a refusal: the slot is released on failure") | |||
| 238 | + | } | |||
| 239 | + | } | |||
| 240 | + | ||||
| 241 | + | // deadlineRecorder is a ResponseWriter that supports the read deadline, as | |||
| 242 | + | // the writer of net/http does, and records it. | |||
| 243 | + | type deadlineRecorder struct { | |||
| 244 | + | *httptest.ResponseRecorder | |||
| 245 | + | deadlines []time.Time | |||
| 246 | + | } | |||
| 247 | + | ||||
| 248 | + | func (d *deadlineRecorder) SetReadDeadline(t time.Time) error { | |||
| 249 | + | d.deadlines = append(d.deadlines, t) | |||
| 250 | + | return nil | |||
| 251 | + | } | |||
| 252 | + | ||||
| 253 | + | func TestHandlerBodyReadDeadline(t *testing.T) { | |||
| 254 | + | tests := []struct { | |||
| 255 | + | name string | |||
| 256 | + | path string | |||
| 257 | + | max time.Duration | |||
| 258 | + | min time.Duration | |||
| 259 | + | }{ | |||
| 260 | + | {"fetch body", pathUpload, time.Minute, 30 * time.Second}, | |||
| 261 | + | {"push body is bounded by idleness", pathReceive, 5 * time.Minute, time.Minute}, | |||
| 262 | + | } | |||
| 263 | + | for _, tt := range tests { | |||
| 264 | + | t.Run(tt.name, func(t *testing.T) { | |||
| 265 | + | r := newRig(t, nil) | |||
| 266 | + | req := httptest.NewRequest(http.MethodPost, tt.path, stringsReader("payload")) | |||
| 267 | + | req.SetBasicAuth("u", "root-secret") | |||
| 268 | + | w := &deadlineRecorder{ResponseRecorder: httptest.NewRecorder()} | |||
| 269 | + | start := time.Now() | |||
| 270 | + | r.h.ServeHTTP(w, req) | |||
| 271 | + | require.Equal(t, http.StatusOK, w.Code) | |||
| 272 | + | ||||
| 273 | + | require.GreaterOrEqual(t, len(w.deadlines), 2) | |||
| 274 | + | first, last := w.deadlines[0], w.deadlines[len(w.deadlines)-1] | |||
| 275 | + | assert.WithinRange(t, first, start.Add(tt.min), time.Now().Add(tt.max)) | |||
| 276 | + | assert.True(t, last.IsZero(), "the deadline is lifted once the body is read") | |||
| 277 | + | }) | |||
| 278 | + | } | |||
| 279 | + | } | |||
| 280 | + | ||||
| 281 | + | func stringsReader(s string) *strings.Reader { return strings.NewReader(s) } | |||