Bound concurrent /api/v1/... requests and cache the history response

SECURITY.md

@@ -69,6 +69,14 @@ reporters and reviewers have context:
   anything beyond `PATH` (and, for `apt`, a couple of apt-specific
   variables), so `PIMONITOR_API_KEY` is never copied into a child process's
   `/proc/<pid>/environ`.
+- **`/api/v1/...` requests share a concurrency limit** (`withMaxInFlight`,
+  `internal/httpapi/middleware.go`), so a request flood — from another
+  device on the LAN, a misconfigured integration, or a buggy retry loop —
+  is bounded rather than able to multiply CPU/memory work without limit on
+  constrained hardware such as a Pi Zero. This does not require
+  authentication and is not a substitute for `api_key`; it is a resource-
+  exhaustion mitigation, not an access control. See
+  [`docs/API.md`](docs/API.md#rate-limiting).
 
 If you believe any of these assumptions are violated by the current
 implementation, please report it as described above.

docs/API.md

@@ -45,6 +45,20 @@ and `Vary: Accept-Encoding`); the JSON body is unchanged, only its wire
 encoding differs. Requests without that header receive the identity
 (uncompressed) response, so existing clients keep working unmodified.
 
+## Rate limiting
+
+Every `/api/v1/...` endpoint shares a single limit on how many requests may be actively
+processing at once. `GET /api/v1/metrics/history` is the expensive one — it can require
+copying and re-serialising the whole retained history window — so an unbounded number of
+concurrent callers could otherwise starve metric collection on constrained hardware such
+as a Raspberry Pi Zero.
+
+A request beyond the limit receives `503 Service Unavailable` with a `Retry-After: 1`
+header and a plain-text body, instead of being queued. Clients should treat this the same
+as any other transient server error: back off (the `Retry-After` value is in seconds) and
+retry. `GET /healthz` is never subject to this limit, so liveness checks keep working even
+while the API is shedding load.
+
 ## Endpoints
 
 ### `GET /healthz`

docs/ARCHITECTURE.md

@@ -242,17 +242,32 @@ withLogging(withSecurityHeaders(mux))
   or the configured key's *length* through response timing. When `APIKey` is empty (the
   default), every request is allowed — the common case (dashboard on a trusted LAN) stays
   unauthenticated and simple.
+- **`withMaxInFlight`**: wraps each of the four `/api/v1/...` routes (not `/healthz` or
+  the static dashboard) in a shared semaphore of capacity `defaultMaxInFlight` (16). A
+  request beyond the limit gets `503 Service Unavailable` with `Retry-After: 1` instead of
+  being queued indefinitely. This bounds worst-case concurrent CPU/memory on a Pi Zero: an
+  unauthenticated flood of requests to the history endpoint (the most expensive one — see
+  below) can no longer multiply without limit. `/healthz` is deliberately excluded so a
+  monitoring system can still tell the process is alive while the API is shedding load.
 
 **Routes**: `GET /healthz` (plain-text liveness, never gated), and four versioned,
 API-key-gated routes — `GET /api/v1/metrics`, `GET /api/v1/metrics/history`,
 `GET /api/v1/alerts`, `GET /api/v1/config` — plus `GET /` serving the embedded dashboard
 via the `staticHandler` passed into `New` (`nil` in tests, to exercise the API layer
 without the frontend). See [`API.md`](API.md) for the full response schemas.
 
-**`MetricsProvider` is a narrow interface** (`Snapshot() / History() / Alerts()`)
-implemented by `*collector.Collector`, so `httpapi` can be unit-tested against a fake
-implementation (see `handlers_test.go`) entirely independent of real `/proc`/`/sys`
-access — the same testability principle the collector package applies to its own parsers.
+**`MetricsProvider` is a narrow interface** (`Snapshot() / History() / HistoryGeneration()
+/ Alerts()`) implemented by `*collector.Collector`, so `httpapi` can be unit-tested
+against a fake implementation (see `handlers_test.go`) entirely independent of real
+`/proc`/`/sys` access — the same testability principle the collector package applies to
+its own parsers.
+
+**`handleHistory` caches its encoded response.** `Collector.History()` deep-copies every
+retained ring buffer under lock — the most expensive request the service serves — but the
+underlying data only changes once per `fastTick`. `Collector.HistoryGeneration()` returns
+a counter bumped each `fastTick`; `Server` keeps the last generation it encoded alongside
+the encoded bytes and reuses them whenever the generation is unchanged, so concurrent
+pollers between ticks share one deep-copy-and-encode instead of each paying for their own.
 
 **`GET /api/v1/config`** exists specifically so the frontend doesn't have to duplicate
 values (poll interval, alert thresholds, feature toggles) that are already defined

internal/collector/collector.go

@@ -106,18 +106,24 @@ type Collector struct {
 	// directly).
 	fastInterval time.Duration
 
-	mu       sync.RWMutex
-	latest   Snapshot
-	cpuHist  *RingBuffer[HistoryPoint]
-	l1Hist   *RingBuffer[HistoryPoint]
-	l5Hist   *RingBuffer[HistoryPoint]
-	l15Hist  *RingBuffer[HistoryPoint]
-	tempHist *RingBuffer[HistoryPoint]
-	memHist  *RingBuffer[HistoryPoint]
-	swapHist *RingBuffer[HistoryPoint]
-	diskHist map[string]*RingBuffer[HistoryPoint]
-	rxHist   map[string]*RingBuffer[HistoryPoint]
-	txHist   map[string]*RingBuffer[HistoryPoint]
+	mu     sync.RWMutex
+	latest Snapshot
+	// historyGen counts fastTicks that have recorded at least one history
+	// point. HTTP handlers use it to detect whether the retained history has
+	// actually changed since a cached, already-serialised response was built,
+	// so a client polling faster than the collector ticks doesn't force a
+	// fresh deep-copy-and-encode of the whole window on every request.
+	historyGen uint64
+	cpuHist    *RingBuffer[HistoryPoint]
+	l1Hist     *RingBuffer[HistoryPoint]
+	l5Hist     *RingBuffer[HistoryPoint]
+	l15Hist    *RingBuffer[HistoryPoint]
+	tempHist   *RingBuffer[HistoryPoint]
+	memHist    *RingBuffer[HistoryPoint]
+	swapHist   *RingBuffer[HistoryPoint]
+	diskHist   map[string]*RingBuffer[HistoryPoint]
+	rxHist     map[string]*RingBuffer[HistoryPoint]
+	txHist     map[string]*RingBuffer[HistoryPoint]
 
 	// persistWG tracks in-flight persistHistory writes so Run's ctx.Done()
 	// branch can wait for the final flush before returning.
@@ -284,6 +290,16 @@ func (c *Collector) History() History {
 	return h
 }
 
+// HistoryGeneration returns a counter that increments every time fastTick
+// records a new set of history points. Callers can compare successive
+// values to tell whether History() would return anything different without
+// having to call it (which deep-copies every ring buffer).
+func (c *Collector) HistoryGeneration() uint64 {
+	c.mu.RLock()
+	defer c.mu.RUnlock()
+	return c.historyGen
+}
+
 func (c *Collector) collectSysInfo() {
 	info := c.sysInfo.Collect()
 	if !c.cfg.DistroInfoEnabled {
@@ -409,6 +425,7 @@ func (c *Collector) fastTick(ctx context.Context) {
 	c.swapHist.Add(HistoryPoint{Timestamp: s.now, Value: s.swap.UsedPercent})
 
 	c.recordDeviceHistory(s.now, s.disks, s.netIfaces)
+	c.historyGen++
 
 	// Evaluate the freshly collected values against the alert thresholds.
 	// The engine has its own lock and never calls back into the collector,

internal/collector/collector_test.go

@@ -120,6 +120,25 @@ func TestCollector_FastTick_BuildsHistory(t *testing.T) {
 	}
 }
 
+func TestCollector_HistoryGeneration_IncrementsOnFastTick(t *testing.T) {
+	c := newTestCollector()
+	ctx := context.Background()
+
+	if got := c.HistoryGeneration(); got != 0 {
+		t.Fatalf("HistoryGeneration before any tick = %d, want 0", got)
+	}
+
+	c.fastTick(ctx)
+	if got := c.HistoryGeneration(); got != 1 {
+		t.Fatalf("HistoryGeneration after 1 tick = %d, want 1", got)
+	}
+
+	c.fastTick(ctx)
+	if got := c.HistoryGeneration(); got != 2 {
+		t.Fatalf("HistoryGeneration after 2 ticks = %d, want 2", got)
+	}
+}
+
 func TestCollector_CollectSysInfo_TogglesDistroAndPiModel(t *testing.T) {
 	c := New(Config{
 		FastInterval:      time.Second,

internal/httpapi/handlers.go

@@ -5,15 +5,17 @@ import (
 	"net/http"
 )
 
+const contentTypeHeader = "Content-Type"
+
 func writeJSON(w http.ResponseWriter, log func(msg string, args ...any), v any) {
-	w.Header().Set("Content-Type", "application/json; charset=utf-8")
+	w.Header().Set(contentTypeHeader, "application/json; charset=utf-8")
 	if err := json.NewEncoder(w).Encode(v); err != nil {
 		log("failed to encode JSON response", "error", err)
 	}
 }
 
 func (s *Server) handleHealthz(w http.ResponseWriter, _ *http.Request) {
-	w.Header().Set("Content-Type", "text/plain; charset=utf-8")
+	w.Header().Set(contentTypeHeader, "text/plain; charset=utf-8")
 	_, _ = w.Write([]byte("ok"))
 }
 
@@ -26,9 +28,31 @@ func (s *Server) handleMetrics(w http.ResponseWriter, _ *http.Request) {
 }
 
 // handleHistory serves GET /api/v1/metrics/history: the retained
-// in-memory history for every time-series metric.
+// in-memory history for every time-series metric. The retained history only
+// changes once per fastTick, so the encoded response is cached and reused
+// across requests until MetricsProvider.HistoryGeneration() moves on,
+// rather than deep-copying every ring buffer and re-encoding on every
+// request from every dashboard/integration poll.
 func (s *Server) handleHistory(w http.ResponseWriter, _ *http.Request) {
-	writeJSON(w, s.log.Error, s.metrics.History())
+	gen := s.metrics.HistoryGeneration()
+
+	s.historyCacheMu.Lock()
+	if s.historyCacheJSON == nil || s.historyCacheGen != gen {
+		data, err := json.Marshal(s.metrics.History())
+		if err != nil {
+			s.historyCacheMu.Unlock()
+			s.log.Error("failed to encode JSON response", "error", err)
+			http.Error(w, "internal error", http.StatusInternalServerError)
+			return
+		}
+		s.historyCacheJSON = data
+		s.historyCacheGen = gen
+	}
+	data := s.historyCacheJSON
+	s.historyCacheMu.Unlock()
+
+	w.Header().Set(contentTypeHeader, "application/json; charset=utf-8")
+	_, _ = w.Write(data)
 }
 
 // handleAlerts serves GET /api/v1/alerts: the current per-metric alert

internal/httpapi/handlers_test.go

@@ -2,6 +2,7 @@ package httpapi
 
 import (
 	"encoding/json"
+	"math"
 	"net/http"
 	"net/http/httptest"
 	"testing"
@@ -14,12 +15,23 @@ import (
 type fakeMetrics struct {
 	snapshot collector.Snapshot
 	history  collector.History
-	alerts   alert.Report
+	// historyGen is returned verbatim by HistoryGeneration, so tests can
+	// simulate a fastTick by incrementing it.
+	historyGen uint64
+	// historyCalls counts calls to History(), so tests can assert the
+	// generation-based cache in handleHistory actually skips re-encoding
+	// when the generation hasn't moved.
+	historyCalls int
+	alerts       alert.Report
 }
 
 func (f *fakeMetrics) Snapshot() collector.Snapshot { return f.snapshot }
-func (f *fakeMetrics) History() collector.History   { return f.history }
-func (f *fakeMetrics) Alerts() alert.Report         { return f.alerts }
+func (f *fakeMetrics) History() collector.History {
+	f.historyCalls++
+	return f.history
+}
+func (f *fakeMetrics) HistoryGeneration() uint64 { return f.historyGen }
+func (f *fakeMetrics) Alerts() alert.Report      { return f.alerts }
 
 func newTestServer(cfg Config) (*Server, *fakeMetrics) {
 	fm := &fakeMetrics{
@@ -88,6 +100,80 @@ func TestHandleHistory(t *testing.T) {
 	}
 }
 
+// TestHandleHistory_CachesUntilGenerationChanges is the regression test for
+// the generation-based response cache: repeated requests while
+// HistoryGeneration() is unchanged must reuse the cached encoding rather
+// than calling History() (and therefore deep-copying every ring buffer)
+// again, but a request after the generation advances must see fresh data.
+func TestHandleHistory_CachesUntilGenerationChanges(t *testing.T) {
+	s, fm := newTestServer(Config{})
+
+	req := func() *httptest.ResponseRecorder {
+		rec := httptest.NewRecorder()
+		s.Handler().ServeHTTP(rec, httptest.NewRequest(http.MethodGet, "/api/v1/metrics/history", nil))
+		return rec
+	}
+
+	first := req()
+	if first.Code != http.StatusOK {
+		t.Fatalf("first request status = %d, want 200", first.Code)
+	}
+	if fm.historyCalls != 1 {
+		t.Fatalf("historyCalls after first request = %d, want 1", fm.historyCalls)
+	}
+
+	second := req()
+	if second.Code != http.StatusOK {
+		t.Fatalf("second request status = %d, want 200", second.Code)
+	}
+	if fm.historyCalls != 1 {
+		t.Fatalf("historyCalls after second request (same generation) = %d, want still 1 (cache should have been reused)", fm.historyCalls)
+	}
+	if second.Body.String() != first.Body.String() {
+		t.Fatalf("cached response body changed between requests with the same generation:\nfirst:  %s\nsecond: %s", first.Body.String(), second.Body.String())
+	}
+
+	fm.history.CPUPercent = append(fm.history.CPUPercent, collector.HistoryPoint{Value: 99})
+	fm.historyGen++
+
+	third := req()
+	if third.Code != http.StatusOK {
+		t.Fatalf("third request status = %d, want 200", third.Code)
+	}
+	if fm.historyCalls != 2 {
+		t.Fatalf("historyCalls after generation changed = %d, want 2 (cache should have been invalidated)", fm.historyCalls)
+	}
+	var got collector.History
+	if err := json.Unmarshal(third.Body.Bytes(), &got); err != nil {
+		t.Fatalf("unmarshal third response: %v", err)
+	}
+	if len(got.CPUPercent) != 2 {
+		t.Fatalf("expected 2 CPUPercent history points after generation change, got %d", len(got.CPUPercent))
+	}
+}
+
+// TestHandleHistory_MarshalFailureReturns500 exercises the error branch
+// that skips the cache: json.Marshal fails on a NaN float (not
+// representable in JSON), which handleHistory must turn into a 500 rather
+// than caching a partial/garbage encoding or writing a broken body.
+func TestHandleHistory_MarshalFailureReturns500(t *testing.T) {
+	s, fm := newTestServer(Config{})
+	fm.history = collector.History{
+		CPUPercent: []collector.HistoryPoint{{Value: math.NaN()}},
+	}
+
+	req := httptest.NewRequest(http.MethodGet, "/api/v1/metrics/history", nil)
+	rec := httptest.NewRecorder()
+	s.Handler().ServeHTTP(rec, req)
+
+	if rec.Code != http.StatusInternalServerError {
+		t.Fatalf("status = %d, want %d", rec.Code, http.StatusInternalServerError)
+	}
+	if s.historyCacheJSON != nil {
+		t.Fatal("expected no cached response after a marshal failure")
+	}
+}
+
 func TestHandleAlerts(t *testing.T) {
 	s, fm := newTestServer(Config{})
 	fm.alerts = alert.Report{

internal/httpapi/middleware.go

@@ -68,6 +68,26 @@ func (s *Server) withAPIKey(next http.Handler) http.Handler {
 	})
 }
 
+// withMaxInFlight bounds concurrent request processing across every
+// endpoint sharing s.inFlight (see defaultMaxInFlight). The history
+// endpoint deep-copies and serialises the whole retained window on a cache
+// miss, so an unbounded number of concurrent callers can starve the
+// collector's own tick on a single-core Pi. Excess requests get 503 with
+// Retry-After rather than being queued indefinitely, so a client backs off
+// instead of piling up.
+func (s *Server) withMaxInFlight(next http.Handler) http.Handler {
+	return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		select {
+		case s.inFlight <- struct{}{}:
+			defer func() { <-s.inFlight }()
+			next.ServeHTTP(w, r)
+		default:
+			w.Header().Set("Retry-After", "1")
+			http.Error(w, "server busy", http.StatusServiceUnavailable)
+		}
+	})
+}
+
 func providedAPIKey(r *http.Request) string {
 	if key := r.Header.Get("X-Api-Key"); key != "" {
 		return key