From f1b49d025182f08a17858d3381a1dfaa88cd5294 Mon Sep 17 00:00:00 2001 From: Ilker Manap Date: Sun, 4 Oct 2026 10:21:15 +0200 Subject: [PATCH] Bound idle time, header size and request reading on every listener, and connections on the client port Co-Authored-By: Claude Opus 5.5 (1M context) --- CHANGELOG.md | 7 +++++ docs/reference.md | 8 +++++ internal/agent/engines/unsafeenv.go | 14 ++++----- internal/cli/mirror.go | 20 ++++++------- internal/cli/server.go | 44 ++++++++++++++++++++-------- internal/config/config.go | 12 +++++--- internal/control/agentsvc/service.go | 4 ++- internal/docgen/reference_gen.go | 1 + internal/join/join.go | 6 ++-- 9 files changed, 78 insertions(+), 38 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 4aef7f2..029478e 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -119,6 +119,13 @@ APIs change between minors without a deprecation period. first time. The cache is now bounded, and keys not in it are looked up at most 20 a second per client address (50 at once); beyond that the request is answered 429 without a read. +- **Slow or idle connections cannot pile up.** The servers bounded only how + long headers took. Every listener now also bounds how long a connection + may sit idle and how large its headers may be, and reading a request + (5 minutes on the client port, a minute or less elsewhere — never on the + agents' long-lived streams); the client port holds at most + `max_connections` (10,000) at once, and a message from an agent at most + 16 MB. ### Everything that can be changed on a running cluster, from the console diff --git a/docs/reference.md b/docs/reference.md index a657f37..78bd521 100644 --- a/docs/reference.md +++ b/docs/reference.md @@ -173,6 +173,7 @@ rather than repeated. | `threads` | `Threads` | [Server](#server) | | `landing` | `Landing` | [Server](#server) | | `agent_tls` | `tlsconf.Config` | [Server](#server) | +| `max_connections` | `int` | [Server](#server) | | `admin_token_file` | `string` | [Server](#server) | | `admin_remote` | `bool` | [Server](#server) | | `debug_pprof` | `bool` | [Server](#server) | @@ -1381,6 +1382,13 @@ trusted certificate, while this one is mutual and private. Takes a [tlsconf.Config](#tlsconfconfig) block. +### `max_connections` + +`int` + +MaxConnections bounds the client port's open connections; past it new +ones wait in the kernel's backlog. 0 is no bound. + ### `admin_token_file` `string` diff --git a/internal/agent/engines/unsafeenv.go b/internal/agent/engines/unsafeenv.go index 43ab6af..277b8a8 100644 --- a/internal/agent/engines/unsafeenv.go +++ b/internal/agent/engines/unsafeenv.go @@ -91,10 +91,10 @@ var flagsNever = map[string]map[string]string{ "--path": "serves a directory's files over HTTP", "--rpc": "sends the model's work to other machines", "--chat-template-file": "reads a file", "--system-prompt-file": "reads a file", "--grammar-file": "reads a file", "--json-schema-file": "reads a file", - "--lora": "loads a file; adapters are set in the model's adapters", - "--lora-scaled": "loads a file; adapters are set in the model's adapters", + "--lora": "loads a file; adapters are set in the model's adapters", + "--lora-scaled": "loads a file; adapters are set in the model's adapters", "--control-vector": "loads a file", "--control-vector-scaled": "loads a file", - "--mmproj": "loads a file; the projector is fetched with the model", + "--mmproj": "loads a file; the projector is fetched with the model", "--model-url": "downloads a model", "-mu": "downloads a model", "--hf-repo": "downloads a model", "-hf": "downloads a model", "--hf-file": "downloads a model", "--hf-token": "is a credential", "--alias": "is set by the agent", @@ -104,13 +104,13 @@ var flagsNever = map[string]map[string]string{ "vllm": { "--download-dir": "writes files", "--tokenizer": "reads another repository or path", "--chat-template": "reads a file", "--allowed-local-media-path": "serves local files to requests", - "--lora-modules": "loads files; adapters are set in the model's adapters", + "--lora-modules": "loads files; adapters are set in the model's adapters", "--tool-parser-plugin": "runs a Python file", "--middleware": "imports Python", "--worker-extension-cls": "imports Python", "--worker-cls": "imports Python", - "--load-format": "changes how weights are loaded, some ways running code", + "--load-format": "changes how weights are loaded, some ways running code", "--otlp-traces-endpoint": "sends request traces elsewhere", - "--served-model-name": "is set by the agent", - "--enable-log-requests": "logs prompts", + "--served-model-name": "is set by the agent", + "--enable-log-requests": "logs prompts", }, "sglang": { "--tokenizer-path": "reads another repository or path", "--chat-template": "reads a file", diff --git a/internal/cli/mirror.go b/internal/cli/mirror.go index d43165a..61cdf40 100644 --- a/internal/cli/mirror.go +++ b/internal/cli/mirror.go @@ -293,20 +293,20 @@ func newMirrorServeCmd(dir *string) *cobra.Command { // http.Dir refuses paths that escape the root, which is the whole // traversal question answered by the standard library rather than // by me. - files := &http.Server{ - Addr: listen, - Handler: readOnly(withManifest(*dir, http.FileServer(http.Dir(*dir)))), - ReadHeaderTimeout: 10 * time.Second, - } + files := withLimits(&http.Server{ + Addr: listen, + Handler: readOnly(withManifest(*dir, http.FileServer(http.Dir(*dir)))), + ReadTimeout: time.Minute, + }) allowed, err := parseAllow(allow) if err != nil { return err } - admin := &http.Server{ - Addr: apiListen, - Handler: onlyClients(clients, apiGate(tlsc.Enabled(), allowed, api.Handler())), - ReadHeaderTimeout: 10 * time.Second, - } + admin := withLimits(&http.Server{ + Addr: apiListen, + Handler: onlyClients(clients, apiGate(tlsc.Enabled(), allowed, api.Handler())), + ReadTimeout: time.Minute, + }) errs := make(chan error, 2) go func() { errs <- fmt.Errorf("serving weights: %w", files.ListenAndServe()) }() diff --git a/internal/cli/server.go b/internal/cli/server.go index 3391417..d6fe19d 100644 --- a/internal/cli/server.go +++ b/internal/cli/server.go @@ -26,6 +26,7 @@ import ( "github.com/spf13/cobra" "golang.org/x/net/http2" "golang.org/x/net/http2/h2c" + "golang.org/x/net/netutil" "golang.org/x/sync/errgroup" llmgwv1 "gitea.mynodes.xyz/ilker/llmgw/api/gen/llmgw/v1" @@ -348,12 +349,15 @@ func newServerCmd() *cobra.Command { "api", cfg.Mirror.API, "mtls", cfg.Mirror.TLS.Enabled()) } - apiSrv := &http.Server{ - Handler: rt.Handler(), - ReadHeaderTimeout: 10 * time.Second, + apiSrv := withLimits(&http.Server{ + Handler: rt.Handler(), + // Reading the request: long enough for a body up to + // max_body_bytes on a slow link, short enough that a client + // trickling one cannot hold a connection and its buffer. + ReadTimeout: 5 * time.Minute, // No write timeout: responses stream for minutes and the proxy // enforces its own per-phase deadlines. - } + }) if err := checkAdminListen(cfg); err != nil { return err } @@ -361,15 +365,20 @@ func newServerCmd() *cobra.Command { if err != nil { return err } - adminSrv := &http.Server{ - Handler: router.AdminGuard(router.AdminAuth(adminToken, cfg.DebugPprof, adminMux)), - ReadHeaderTimeout: 10 * time.Second, - } + adminSrv := withLimits(&http.Server{ + Handler: router.AdminGuard(router.AdminAuth(adminToken, cfg.DebugPprof, adminMux)), + ReadTimeout: time.Minute, + }) apiLn, err := net.Listen("tcp", cfg.Listen) if err != nil { return fmt.Errorf("listening on %s: %w", cfg.Listen, err) } + if cfg.MaxConnections > 0 { + // Past this, new connections wait in the kernel's backlog + // rather than each taking a goroutine and its buffers. + apiLn = netutil.LimitListener(apiLn, cfg.MaxConnections) + } adminLn, err := net.Listen("tcp", cfg.AdminListen) if err != nil { _ = apiLn.Close() @@ -402,7 +411,7 @@ func newServerCmd() *cobra.Command { mux.Handle("/", h) h = mux } - metricsSrv = &http.Server{Handler: h, ReadHeaderTimeout: 10 * time.Second} + metricsSrv = withLimits(&http.Server{Handler: h, ReadTimeout: 30 * time.Second}) go func() { if err := metricsSrv.Serve(metricsLn); err != nil && !errors.Is(err, http.ErrServerClosed) { log.Error("metrics listener stopped", "err", err) @@ -423,10 +432,9 @@ func newServerCmd() *cobra.Command { if agentHandler != nil { mux := http.NewServeMux() mux.Handle(agentPath, agentHandler) - agentSrv = &http.Server{ - Handler: mux, - ReadHeaderTimeout: 10 * time.Second, - } + // No read timeout: an agent's stream is one request that + // lasts as long as the agent does. + agentSrv = withLimits(&http.Server{Handler: mux}) if !cfg.AgentTLS.Enabled() { // Without TLS, agents speak plaintext HTTP/2, which Go will // not do unless asked. Nothing then stops a node from @@ -801,3 +809,13 @@ func agentListenAddr(cfg config.Server) (string, error) { } return cfg.AgentListen, nil } + +// withLimits gives a server what every listener here wants: a bound on how +// long the headers may take, how long a connection may sit idle between +// requests, and how large the headers may be. +func withLimits(s *http.Server) *http.Server { + s.ReadHeaderTimeout = 10 * time.Second + s.IdleTimeout = 2 * time.Minute + s.MaxHeaderBytes = 1 << 20 + return s +} diff --git a/internal/config/config.go b/internal/config/config.go index 1892e95..96cfb13 100644 --- a/internal/config/config.go +++ b/internal/config/config.go @@ -65,6 +65,9 @@ type Server struct { // cover the client-facing one: that surface is fronted by a publicly // trusted certificate, while this one is mutual and private. AgentTLS tlsconf.Config `koanf:"agent_tls"` + // MaxConnections bounds the client port's open connections; past it new + // ones wait in the kernel's backlog. 0 is no bound. + MaxConnections int `koanf:"max_connections"` // AdminTokenFile holds the token the admin API's /api and /debug paths // demand. Empty is admin-token in the systemd unit's state directory // (/var/lib/llmgw), or secrets/admin-token beside this file for a server @@ -503,10 +506,11 @@ type Endpoint struct { // Default returns the configuration used when nothing is specified. func Default() Server { return Server{ - Listen: ":8080", - AdminListen: "127.0.0.1:9090", - AgentListen: ":7443", - Mirror: Mirror{UpdateCheck: 6 * time.Hour}, + Listen: ":8080", + AdminListen: "127.0.0.1:9090", + AgentListen: ":7443", + MaxConnections: 10_000, + Mirror: Mirror{UpdateCheck: 6 * time.Hour}, Control: Control{ Interval: 5 * time.Second, BaseBackoff: 10 * time.Second, diff --git a/internal/control/agentsvc/service.go b/internal/control/agentsvc/service.go index 2813955..2cd88c7 100644 --- a/internal/control/agentsvc/service.go +++ b/internal/control/agentsvc/service.go @@ -142,7 +142,9 @@ func New(inv *inventory.Store, status StatusSink, requireIdentity bool, log *slo // The handler is wrapped so the verified client identity reaches the stream: // connect exposes the peer's address but not its certificate. func (s *Service) Handler() (string, http.Handler) { - path, h := llmgwv1connect.NewAgentServiceHandler(s) + // A message is a heartbeat, a report or a log; nothing an agent says + // needs more, and a bound keeps one connection from taking the memory. + path, h := llmgwv1connect.NewAgentServiceHandler(s, connect.WithReadMaxBytes(16<<20)) return path, AuthMiddleware(h) } diff --git a/internal/docgen/reference_gen.go b/internal/docgen/reference_gen.go index dc4a2b0..444d075 100644 --- a/internal/docgen/reference_gen.go +++ b/internal/docgen/reference_gen.go @@ -213,6 +213,7 @@ var Reference = []Block{ {Key: "threads", Type: "Threads", Doc: "Threads keeps conversations and agent memory without API keys, for a\ngateway on one trusted machine. With auth on they live in the auth store,\nper project, and this block is ignored.", Nested: "config.Threads"}, {Key: "landing", Type: "Landing", Doc: "Landing configures the page served at \"/\".", Nested: "config.Landing"}, {Key: "agent_tls", Type: "tlsconf.Config", Doc: "AgentTLS authenticates the fleet-facing port. It deliberately does not\ncover the client-facing one: that surface is fronted by a publicly\ntrusted certificate, while this one is mutual and private.", Nested: "tlsconf.Config"}, + {Key: "max_connections", Type: "int", Doc: "MaxConnections bounds the client port's open connections; past it new\nones wait in the kernel's backlog. 0 is no bound.", Nested: ""}, {Key: "admin_token_file", Type: "string", Doc: "AdminTokenFile holds the token the admin API's /api and /debug paths\ndemand. Empty is admin-token in the systemd unit's state directory\n(/var/lib/llmgw), or secrets/admin-token beside this file for a server\nrun by hand. Made, 0600, when the server starts and it does not exist;\n`sudo llmgw ctl` and the console read it from there.", Nested: ""}, {Key: "admin_remote", Type: "bool", Doc: "AdminRemote allows admin_listen on an address other than loopback. The\ntoken still guards the API, but it then crosses the network in clear.", Nested: ""}, {Key: "debug_pprof", Type: "bool", Doc: "DebugPprof serves Go's profiler on the admin listener, behind the token.\nOff by default: a heap profile holds the API keys requests presented.", Nested: ""}, diff --git a/internal/join/join.go b/internal/join/join.go index d64d002..f290cd6 100644 --- a/internal/join/join.go +++ b/internal/join/join.go @@ -131,10 +131,10 @@ type response struct { // Listener waits for machines to join. type Listener struct { - ln net.Listener - addr string + ln net.Listener + addr string // now is time.Now, replaced by tests. - now func() time.Time + now func() time.Time fingerprint [16]byte ttl time.Duration