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) <noreply@anthropic.com>

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
 

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`

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",

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()) }()

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
+}

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,

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

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: ""},

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