feat(ws): limit connection attempts and signal capacity shedding

SHOULDERS.md

@@ -12,6 +12,8 @@ projects that make this possible:
 - [sqlc](https://sqlc.dev) — Type-safe SQL code generation
 - [chi](https://github.com/go-chi/chi) — HTTP router
 - [go-chi/cors](https://github.com/go-chi/cors) — CORS middleware for chi
+- [go-chi/httprate](https://github.com/go-chi/httprate) — Per-client HTTP
+  sliding-window rate limits and retry headers
 - [go-redis](https://github.com/redis/go-redis) — Redis client
 - [coder/websocket](https://github.com/coder/websocket) — WebSocket
   implementation

cmd/beacon/main.go

@@ -41,6 +41,7 @@ var version = "dev"
 //	@title			MeshCore Beacon API
 //	@version		1.6.0
 //	@description	MeshCore network observation backend. Ingests LoRa packets from MQTT brokers, stores in PostgreSQL, and streams live events via WebSocket.
+//	@description	WebSocket upgrade attempts at /ws have a configurable per-client limit (default 10/minute, including failed handshakes). Rate exhaustion returns HTTP 429 with Retry-After before upgrade. An accepted socket exceeding the concurrent cap closes with code 1013 before hello; established connections remain open.
 //	@termsOfService	https://github.com/MeshCore-Beacon/beacon-server
 
 //	@contact.name	MeshCore Beacon
@@ -275,7 +276,7 @@ func main() {
 	go scheduler.Start(ctx)
 
 	// ── HTTP server ──────────────────────────────────────────────────────────
-	r := router.New(h, reader, []*ingest.Worker{broker1, broker2}, resolved.MaxConnsPerIP, cfg.CORS, cfg.Server)
+	r := router.New(h, reader, []*ingest.Worker{broker1, broker2}, resolved.MaxConnsPerIP, resolved.MaxConnectsPerMinute, cfg.CORS, cfg.Server)
 
 	srv := &http.Server{
 		Addr:    addr,

config.yaml.example

@@ -115,6 +115,12 @@ routes:
 
 websocket:
   max_connections_per_ip: 5 # default: 5
+  # Upgrade attempts, including failed handshakes; zero/omitted defaults to 10.
+  # IPv6 clients share a /64 attempt budget. Negative values prevent startup.
+  # Rate exhaustion rejects the handshake with 429 and Retry-After: 60.
+  # Within this rate budget, a full concurrent cap accepts then closes with 1013
+  # before hello, allowing browsers to back off without evicting existing clients.
+  max_connects_per_minute: 10
 
 # Node staleness, deletion, and clock-drift thresholds.
 #nodes:

docs/docs.go

@@ -4116,7 +4116,7 @@ var SwaggerInfo = &swag.Spec{
 	BasePath:         "/api/v1",
 	Schemes:          []string{"http", "https"},
 	Title:            "MeshCore Beacon API",
-	Description:      "MeshCore network observation backend. Ingests LoRa packets from MQTT brokers, stores in PostgreSQL, and streams live events via WebSocket.",
+	Description:      "MeshCore network observation backend. Ingests LoRa packets from MQTT brokers, stores in PostgreSQL, and streams live events via WebSocket.\nWebSocket upgrade attempts at /ws have a configurable per-client limit (default 10/minute, including failed handshakes). Rate exhaustion returns HTTP 429 with Retry-After before upgrade. An accepted socket exceeding the concurrent cap closes with code 1013 before hello; established connections remain open.",
 	InfoInstanceName: "swagger",
 	SwaggerTemplate:  docTemplate,
 	LeftDelim:        "{{",

docs/swagger.json

@@ -5,7 +5,7 @@
     ],
     "swagger": "2.0",
     "info": {
-        "description": "MeshCore network observation backend. Ingests LoRa packets from MQTT brokers, stores in PostgreSQL, and streams live events via WebSocket.",
+        "description": "MeshCore network observation backend. Ingests LoRa packets from MQTT brokers, stores in PostgreSQL, and streams live events via WebSocket.\nWebSocket upgrade attempts at /ws have a configurable per-client limit (default 10/minute, including failed handshakes). Rate exhaustion returns HTTP 429 with Retry-After before upgrade. An accepted socket exceeding the concurrent cap closes with code 1013 before hello; established connections remain open.",
         "title": "MeshCore Beacon API",
         "termsOfService": "https://github.com/MeshCore-Beacon/beacon-server",
         "contact": {

docs/swagger.yaml

@@ -1182,8 +1182,9 @@ info:
   contact:
     name: MeshCore Beacon
     url: https://github.com/MeshCore-Beacon/beacon-server
-  description: MeshCore network observation backend. Ingests LoRa packets from MQTT
-    brokers, stores in PostgreSQL, and streams live events via WebSocket.
+  description: |-
+    MeshCore network observation backend. Ingests LoRa packets from MQTT brokers, stores in PostgreSQL, and streams live events via WebSocket.
+    WebSocket upgrade attempts at /ws have a configurable per-client limit (default 10/minute, including failed handshakes). Rate exhaustion returns HTTP 429 with Retry-After before upgrade. An accepted socket exceeding the concurrent cap closes with code 1013 before hello; established connections remain open.
   license:
     name: AGPL-3-or-later
   termsOfService: https://github.com/MeshCore-Beacon/beacon-server

go.mod

@@ -8,6 +8,7 @@ require (
 	github.com/eclipse/paho.mqtt.golang v1.5.1
 	github.com/go-chi/chi/v5 v5.3.0
 	github.com/go-chi/cors v1.2.2
+	github.com/go-chi/httprate v0.16.0
 	github.com/google/uuid v1.6.0
 	github.com/jackc/pgx/v5 v5.9.2
 	github.com/joho/godotenv v1.5.1
@@ -33,16 +34,19 @@ require (
 	github.com/jackc/pgservicefile v0.0.0-20240606120523-5a60cdf6a761 // indirect
 	github.com/jackc/puddle/v2 v2.2.2 // indirect
 	github.com/josharian/intern v1.0.0 // indirect
+	github.com/klauspost/cpuid/v2 v2.2.10 // indirect
 	github.com/mailru/easyjson v0.7.6 // indirect
 	github.com/rogpeppe/go-internal v1.14.1 // indirect
 	github.com/swaggo/files v0.0.0-20220610200504-28940afbdbfe // indirect
 	github.com/yuin/gopher-lua v1.1.1 // indirect
+	github.com/zeebo/xxh3 v1.1.0 // indirect
 	go.mongodb.org/mongo-driver/v2 v2.5.0 // indirect
 	go.uber.org/atomic v1.11.0 // indirect
 	golang.org/x/crypto v0.53.0 // indirect
 	golang.org/x/mod v0.37.0 // indirect
 	golang.org/x/net v0.56.0 // indirect
 	golang.org/x/sync v0.21.0 // indirect
+	golang.org/x/sys v0.46.0 // indirect
 	golang.org/x/text v0.39.0 // indirect
 	golang.org/x/tools v0.47.0 // indirect
 	gopkg.in/yaml.v2 v2.4.0 // indirect

go.sum

@@ -22,6 +22,8 @@ github.com/go-chi/chi/v5 v5.3.0 h1:halUjDxhshgXHMrao5bB8eNBXo/rnzwr8m5m36glehM=
 github.com/go-chi/chi/v5 v5.3.0/go.mod h1:R+tYY2hNuVUUjxoPtqUdgBqevM9s9njzkTLutVsOCto=
 github.com/go-chi/cors v1.2.2 h1:Jmey33TE+b+rB7fT8MUy1u0I4L+NARQlK6LhzKPSyQE=
 github.com/go-chi/cors v1.2.2/go.mod h1:sSbTewc+6wYHBBCW7ytsFSn836hqM7JxpglAy2Vzc58=
+github.com/go-chi/httprate v0.16.0 h1:8V5DH9j6pSK6UQoBsTpvMyFxycqaKEIToyPKzHJjUa8=
+github.com/go-chi/httprate v0.16.0/go.mod h1:A8lo+qRhk+s9LiuP5saS7XCGDXRXMcrueq0NfIuCa/I=
 github.com/go-openapi/jsonpointer v0.19.3/go.mod h1:Pl9vOtqEWErmShwVjC8pYs9cog34VGT37dQOVbmoatg=
 github.com/go-openapi/jsonpointer v0.19.5 h1:gZr+CIYByUqjcgeLXnQu2gHYQC9o73G2XUeOFYEICuY=
 github.com/go-openapi/jsonpointer v0.19.5/go.mod h1:Pl9vOtqEWErmShwVjC8pYs9cog34VGT37dQOVbmoatg=
@@ -88,6 +90,8 @@ github.com/swaggo/swag v1.16.6 h1:qBNcx53ZaX+M5dxVyTrgQ0PJ/ACK+NzhwcbieTt+9yI=
 github.com/swaggo/swag v1.16.6/go.mod h1:ngP2etMK5a0P3QBizic5MEwpRmluJZPHjXcMoj4Xesg=
 github.com/yuin/gopher-lua v1.1.1 h1:kYKnWBjvbNP4XLT3+bPEwAXJx262OhaHDWDVOPjL46M=
 github.com/yuin/gopher-lua v1.1.1/go.mod h1:GBR0iDaNXjAgGg9zfCvksxSRnQx76gclCIb7kdAd1Pw=
+github.com/zeebo/assert v1.3.0 h1:g7C04CbJuIDKNPFHmsk4hwZDO5O+kntRxzaUoNXj+IQ=
+github.com/zeebo/assert v1.3.0/go.mod h1:Pq9JiuJQpG8JLJdtkwrJESF0Foym2/D9XMU5ciN/wJ0=
 github.com/zeebo/xxh3 v1.1.0 h1:s7DLGDK45Dyfg7++yxI0khrfwq9661w9EN78eP/UZVs=
 github.com/zeebo/xxh3 v1.1.0/go.mod h1:IisAie1LELR4xhVinxWS5+zf1lA4p0MW4T+w+W07F5s=
 go.mongodb.org/mongo-driver/v2 v2.5.0 h1:yXUhImUjjAInNcpTcAlPHiT7bIXhshCTL3jVBkF3xaE=

internal/api/router/router.go

@@ -41,7 +41,7 @@ import (
 //
 // The private group is stubbed and ready for the auth middleware drop-in
 // described in Future Features → Admin authentication.
-func New(h *hub.Hub, reader api.Reader, workers []*ingest.Worker, maxConnsPerIP int, corsCfg config.CORSConfig, serverCfg config.ServerConfig) http.Handler {
+func New(h *hub.Hub, reader api.Reader, workers []*ingest.Worker, maxConnsPerIP, maxConnectsPerMinute int, corsCfg config.CORSConfig, serverCfg config.ServerConfig) http.Handler {
 	r := chi.NewRouter()
 
 	// ── CORS ─────────────────────────────────────────────────────────────────
@@ -87,7 +87,7 @@ func New(h *hub.Hub, reader api.Reader, workers []*ingest.Worker, maxConnsPerIP
 	))
 
 	// ── WebSocket ────────────────────────────────────────────────────────────
-	r.Get("/ws", ws.Handler(h, reader, maxConnsPerIP))
+	r.Get("/ws", ws.Handler(h, reader, maxConnsPerIP, maxConnectsPerMinute))
 
 	// ── Public REST API (v1) ─────────────────────────────────────────────────
 	r.Route("/api/v1", func(r chi.Router) {

internal/api/router/router_test.go

@@ -19,24 +19,24 @@ import (
 )
 
 func TestWebSocketLimitIgnoresUntrustedHeaders(t *testing.T) {
-	checkWebSocketLimit(t, config.ServerConfig{}, "198.51.100.2", http.StatusTooManyRequests)
+	checkWebSocketLimit(t, config.ServerConfig{}, "198.51.100.2", true)
 }
 
 func TestWebSocketLimitUsesTrustedClientIP(t *testing.T) {
 	cfg := config.ServerConfig{TrustedProxies: []netip.Prefix{
 		netip.MustParsePrefix("127.0.0.1/8"), netip.MustParsePrefix("::1/128"),
 	}}
 	t.Run("different clients", func(t *testing.T) {
-		checkWebSocketLimit(t, cfg, "198.51.100.2", http.StatusSwitchingProtocols)
+		checkWebSocketLimit(t, cfg, "198.51.100.2", false)
 	})
 	t.Run("same client", func(t *testing.T) {
-		checkWebSocketLimit(t, cfg, "198.51.100.1", http.StatusTooManyRequests)
+		checkWebSocketLimit(t, cfg, "198.51.100.1", true)
 	})
 }
 
-func checkWebSocketLimit(t *testing.T, cfg config.ServerConfig, secondIP string, wantStatus int) {
+func checkWebSocketLimit(t *testing.T, cfg config.ServerConfig, secondIP string, wantShed bool) {
 	t.Helper()
-	server := httptest.NewServer(New(hub.New(), nil, nil, 1, config.CORSConfig{}, cfg))
+	server := httptest.NewServer(New(hub.New(), nil, nil, 1, 10, config.CORSConfig{}, cfg))
 	defer server.Close()
 	ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
 	defer cancel()
@@ -55,10 +55,22 @@ func checkWebSocketLimit(t *testing.T, cfg config.ServerConfig, secondIP string,
 		t.Fatalf("read hello: %v", err)
 	}
 	second, response, err := dial(secondIP)
-	if second != nil {
-		second.CloseNow()
+	if err != nil || response == nil || response.StatusCode != http.StatusSwitchingProtocols {
+		t.Fatalf("second handshake: response=%v err=%v", response, err)
 	}
-	if response == nil || response.StatusCode != wantStatus || (err == nil) != (wantStatus == http.StatusSwitchingProtocols) {
-		t.Fatalf("second connection: response=%v err=%v, want status %d", response, err, wantStatus)
+	defer second.CloseNow()
+	_, _, err = second.Read(ctx)
+	if wantShed && websocket.CloseStatus(err) != websocket.StatusTryAgainLater {
+		t.Fatalf("expected close 1013, got %v", err)
+	}
+	if !wantShed && err != nil {
+		t.Fatalf("independent client did not receive hello: %v", err)
+	}
+	if err := first.Write(ctx, websocket.MessageText, []byte(`{"v":1,"type":"ping","id":"still-active"}`)); err != nil {
+		t.Fatal(err)
+	}
+	_, pong, err := first.Read(ctx)
+	if err != nil || !strings.Contains(string(pong), `"type":"pong"`) {
+		t.Fatalf("established client stopped responding: %s %v", pong, err)
 	}
 }

internal/api/router/ws_attempt_test.go

@@ -0,0 +1,53 @@
+// Copyright 2026 Beacon Contributors
+// SPDX-License-Identifier: AGPL-3.0-or-later
+
+package router
+
+import (
+	"net/http"
+	"net/http/httptest"
+	"net/netip"
+	"testing"
+
+	"github.com/MeshCore-Beacon/beacon-server/internal/config"
+)
+
+func TestWebSocketAttemptClientIdentity(t *testing.T) {
+	for _, tc := range []struct {
+		name, firstPeer, secondPeer, firstHeader, secondHeader string
+		trusted, shared                                        bool
+	}{
+		{"untrusted rotation", "198.51.100.1:1", "198.51.100.1:2", "192.0.2.1", "192.0.2.2", false, true},
+		{"independent peers", "198.51.100.1:1", "198.51.100.2:1", "192.0.2.1", "192.0.2.1", false, false},
+		{"trusted independent clients", "127.0.0.1:1", "127.0.0.1:2", "198.51.100.1", "198.51.100.2", true, false},
+		{"trusted shared client", "127.0.0.1:1", "127.0.0.1:2", "198.51.100.1", "198.51.100.1", true, true},
+		{"IPv6 shared prefix", "[2001:db8:1::1]:1", "[2001:db8:1::2]:2", "", "", false, true},
+		{"IPv6 independent prefixes", "[2001:db8:1::1]:1", "[2001:db8:2::1]:2", "", "", false, false},
+	} {
+		t.Run(tc.name, func(t *testing.T) {
+			cfg := config.ServerConfig{}
+			if tc.trusted {
+				cfg.TrustedProxies = []netip.Prefix{netip.MustParsePrefix("127.0.0.1/32")}
+			}
+			handler := New(nil, nil, nil, 5, 1, config.CORSConfig{}, cfg)
+			attempt := func(peer, header string) int {
+				request := httptest.NewRequest(http.MethodGet, "/ws", nil)
+				request.RemoteAddr = peer
+				request.Header.Set("X-Real-IP", header)
+				request.Header.Set("X-Forwarded-For", header)
+				request.Header.Set("True-Client-IP", header)
+				response := httptest.NewRecorder()
+				handler.ServeHTTP(response, request)
+				return response.Code
+			}
+			first := attempt(tc.firstPeer, tc.firstHeader)
+			if first < 400 || first == http.StatusTooManyRequests {
+				t.Fatalf("first failed handshake: %d", first)
+			}
+			second := attempt(tc.secondPeer, tc.secondHeader)
+			if (second == http.StatusTooManyRequests) != tc.shared || second < 400 {
+				t.Fatalf("second handshake status %d; shared budget=%v", second, tc.shared)
+			}
+		})
+	}
+}

internal/config/config.go

@@ -69,6 +69,7 @@ type ResolvedConfig struct {
 	RouteGrace           time.Duration
 	RouteMinObservations int
 	MaxConnsPerIP        int
+	MaxConnectsPerMinute int
 	ViewRefreshInterval  time.Duration
 	ReconfirmInterval    time.Duration
 	CleanupInterval      time.Duration
@@ -201,6 +202,9 @@ type WebSocketConfig struct {
 	// MaxConnectionsPerIP is the maximum number of concurrent WebSocket
 	// connections allowed from a single IP address. Defaults to 5 if not set.
 	MaxConnectionsPerIP int `yaml:"max_connections_per_ip"`
+	// MaxConnectsPerMinute limits upgrade attempts, including failed handshakes.
+	// Zero/omitted defaults to 10; IPv6 addresses share a /64 attempt budget.
+	MaxConnectsPerMinute int `yaml:"max_connects_per_minute"`
 }
 
 // PacketsConfig controls packet retention behaviour.
@@ -345,6 +349,9 @@ func Load(path string) (*Config, error) {
 			return nil, fmt.Errorf("server.trusted_proxies[%d] must be a valid CIDR", i)
 		}
 	}
+	if cfg.WebSocket.MaxConnectsPerMinute < 0 {
+		return nil, fmt.Errorf("websocket.max_connects_per_minute must be positive or zero for the default")
+	}
 	configDir := filepath.Dir(path)
 	for iata, details := range cfg.IATAs {
 		if details.BorderFile != "" && !filepath.IsAbs(details.BorderFile) {
@@ -365,6 +372,7 @@ func Resolve(cfg *Config) ResolvedConfig {
 		RouteGrace:           cfg.Routes.Grace.Duration,
 		RouteMinObservations: cfg.Routes.MinObservations,
 		MaxConnsPerIP:        cfg.WebSocket.MaxConnectionsPerIP,
+		MaxConnectsPerMinute: cfg.WebSocket.MaxConnectsPerMinute,
 		ViewRefreshInterval:  cfg.Background.ViewRefresh.Duration,
 		ReconfirmInterval:    cfg.Background.Reconfirm.Duration,
 		CleanupInterval:      cfg.Background.Cleanup.Duration,
@@ -398,6 +406,9 @@ func Resolve(cfg *Config) ResolvedConfig {
 	if r.MaxConnsPerIP == 0 {
 		r.MaxConnsPerIP = 5
 	}
+	if r.MaxConnectsPerMinute == 0 {
+		r.MaxConnectsPerMinute = 10
+	}
 	if r.ViewRefreshInterval == 0 {
 		r.ViewRefreshInterval = time.Hour
 	}
@@ -429,9 +440,9 @@ func Resolve(cfg *Config) ResolvedConfig {
 
 func (r ResolvedConfig) String() string {
 	return fmt.Sprintf(
-		"telemetryResolution=%s telemetryRetention=%s packetRetention=%s routeRetention=%s routeGrace=%s routeMinObs=%d maxConnsPerIP=%d viewRefresh=%s reconfirm=%s cleanup=%s presenceFlush=%s presencePacketTTL=%s clockDriftThreshold=%s nodeStaleThreshold=%s nodeDeleteAfter=%s observerDeleteAfter=%s",
+		"telemetryResolution=%s telemetryRetention=%s packetRetention=%s routeRetention=%s routeGrace=%s routeMinObs=%d maxConnsPerIP=%d maxConnectsPerMinute=%d viewRefresh=%s reconfirm=%s cleanup=%s presenceFlush=%s presencePacketTTL=%s clockDriftThreshold=%s nodeStaleThreshold=%s nodeDeleteAfter=%s observerDeleteAfter=%s",
 		r.TelemetryResolution, r.TelemetryRetention, r.PacketRetention, r.RouteRetention, r.RouteGrace, r.RouteMinObservations,
-		r.MaxConnsPerIP, r.ViewRefreshInterval, r.ReconfirmInterval, r.CleanupInterval,
+		r.MaxConnsPerIP, r.MaxConnectsPerMinute, r.ViewRefreshInterval, r.ReconfirmInterval, r.CleanupInterval,
 		r.PresenceFlushInterval, r.PresencePacketTTL, r.ClockDriftThreshold,
 		r.NodeStaleThreshold, r.NodeDeleteAfter, r.ObserverDeleteAfter,
 	)

internal/config/ws_connect_test.go

@@ -0,0 +1,37 @@
+// Copyright 2026 Beacon Contributors
+// SPDX-License-Identifier: AGPL-3.0-or-later
+
+package config
+
+import (
+	"os"
+	"path/filepath"
+	"testing"
+)
+
+func TestWebSocketConnectConfig(t *testing.T) {
+	for _, tc := range []struct {
+		name, yaml string
+		want       int
+		wantError  bool
+	}{
+		{"omitted", "{}", 10, false},
+		{"zero default", "websocket: {max_connects_per_minute: 0}", 10, false},
+		{"custom", "websocket: {max_connects_per_minute: 30}", 30, false},
+		{"negative", "websocket: {max_connects_per_minute: -1}", 0, true},
+	} {
+		t.Run(tc.name, func(t *testing.T) {
+			path := filepath.Join(t.TempDir(), "config.yaml")
+			if err := os.WriteFile(path, []byte(tc.yaml), 0600); err != nil {
+				t.Fatal(err)
+			}
+			cfg, err := Load(path)
+			if (err != nil) != tc.wantError {
+				t.Fatalf("Load error = %v, want error = %v", err, tc.wantError)
+			}
+			if err == nil && (Resolve(cfg).MaxConnectsPerMinute != tc.want || Resolve(cfg).MaxConnsPerIP != 5) {
+				t.Fatalf("unexpected resolved WebSocket limits: %+v", Resolve(cfg))
+			}
+		})
+	}
+}