Top wears the skin; the refresh rate reaches every machine; bug bash 4

The process table's columns fit the box they are in, the rate control saves, curves are
timed on the viewer's clock and keep to the stride, the power icon is drawn, and a young
history draws in from the right instead of being stretched.

Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com>

internal/api/server.go

@@ -98,6 +98,24 @@ func New(cfg *config.Config, sessions *auth.Sessions, mgr *term.Manager, col *me
 	if histSec > 0 {
 		col.SetWindow(time.Duration(histSec) * time.Second)
 	}
+	// The sampler follows the console's refresh rate. A machine remembers what it was last told;
+	// the hub reads the rate from the saved UI prefs and tells everyone again, because a node
+	// that restarted, or was added since, would otherwise sample at its default until somebody
+	// happened to open its Dashboard.
+	var samplerMs int
+	var uiRaw json.RawMessage
+	st.View(func(x *store.State) { samplerMs, uiRaw = x.SamplerMs, x.UI })
+	if samplerMs > 0 {
+		col.SetInterval(time.Duration(samplerMs) * time.Millisecond)
+	}
+	if nm != nil {
+		var ui struct {
+			RefreshMs float64 `json:"refreshMs"`
+		}
+		if json.Unmarshal(uiRaw, &ui) == nil && ui.RefreshMs >= 100 {
+			go s.setSampler(int(ui.RefreshMs), true)
+		}
+	}
 	// A node that restarted, or was added since, has its own idea of the window: tell them all.
 	if histSec > 0 && nm != nil {
 		go s.pushHistoryWindow(histSec)
@@ -223,6 +241,8 @@ func New(cfg *config.Config, sessions *auth.Sessions, mgr *term.Manager, col *me
 		r.URL.Path = strings.TrimPrefix(r.URL.Path, "/api")
 		pprof.Index(w, r)
 	}))
+	m.HandleFunc("GET /api/prefs/sampler", s.auth(s.getSampler))
+	m.HandleFunc("PUT /api/prefs/sampler", s.auth(s.putSampler))
 	m.HandleFunc("GET /api/prefs/history", s.auth(s.getHistoryWindow))
 	m.HandleFunc("PUT /api/prefs/history", s.auth(s.putHistoryWindow))
 	m.HandleFunc("GET /api/prefs/dashboard", s.auth(s.getDashPrefs))
@@ -1895,9 +1915,62 @@ func (s *Server) putUIPrefs(w http.ResponseWriter, r *http.Request) {
 		jsonErr(w, 500, err.Error())
 		return
 	}
+	if p.RefreshMs != nil { // the refresh rate is also how often every machine samples
+		go s.setSampler(int(*p.RefreshMs), true)
+	}
 	w.WriteHeader(204)
 }
 
+// ---- how often the machine samples ----
+//
+// One rate for the fleet, the console's refresh rate: set in Settings (or the Top tab), saved
+// with the UI prefs on the hub, and passed by the hub to every machine. Before this the
+// sampler was told only by an open Dashboard page, so a machine whose Dashboard nobody had
+// opened since the rate changed kept sampling at the old one, and every other page of it —
+// Top, Processes, Disks, widgets — "ignored" the setting.
+
+// setSampler applies a rate here and, from a hub, everywhere. The collector's own limits are
+// 100 ms to 10 s; a slower console rate samples at 10 s.
+func (s *Server) setSampler(ms int, fanOut bool) {
+	lo, hi := int(metrics.MinInterval.Milliseconds()), int(metrics.MaxInterval.Milliseconds())
+	if ms < lo {
+		ms = lo
+	}
+	if ms > hi {
+		ms = hi
+	}
+	s.metrics.SetInterval(time.Duration(ms) * time.Millisecond)
+	_ = s.store.Update(func(st *store.State) { st.SamplerMs = ms })
+	if !fanOut || s.nodes == nil {
+		return
+	}
+	ctx, cancel := context.WithTimeout(context.Background(), 20*time.Second)
+	defer cancel()
+	body, _ := json.Marshal(map[string]int{"ms": ms})
+	hdr := http.Header{"X-Hamlab-From-Hub": []string{"1"}, "Content-Type": []string{"application/json"}}
+	for _, name := range s.nodes.Names() {
+		if code, _, err := s.nodes.Do(ctx, name, "PUT", "/api/prefs/sampler", body, hdr); err != nil || code >= 400 {
+			log.Printf("sampler: %s did not take %d ms (%d %v)", name, ms, code, err)
+		}
+	}
+}
+
+func (s *Server) getSampler(w http.ResponseWriter, _ *http.Request) {
+	jsonOK(w, 200, map[string]any{"ms": s.metrics.Interval().Milliseconds(), "min_ms": metrics.MinInterval.Milliseconds(), "max_ms": metrics.MaxInterval.Milliseconds()})
+}
+
+func (s *Server) putSampler(w http.ResponseWriter, r *http.Request) {
+	var body struct {
+		Ms int `json:"ms"`
+	}
+	if json.NewDecoder(io.LimitReader(r.Body, 4096)).Decode(&body) != nil || body.Ms < 100 || body.Ms > 60000 {
+		jsonErr(w, 400, "expected {\"ms\": 100…60000}")
+		return
+	}
+	s.setSampler(body.Ms, r.Header.Get("X-Hamlab-From-Hub") == "")
+	s.getSampler(w, r)
+}
+
 // ---- how far back the graphs reach ----
 //
 // One window for the whole fleet: the hub keeps it and pushes it to every machine, so a

internal/api/status_sensors.go

@@ -37,9 +37,9 @@ const statusHistory = 300
 // A point being averaged, the same idea as the sampler's: these sources are read every 15 to
 // 300 seconds, so a wide window keeps one point per stride rather than one per read.
 type bucket struct {
-	start int64
-	sum   float64
-	n     float64
+	last int64 // when this sensor's last point was kept
+	sum  float64
+	n    float64
 }
 
 const (
@@ -83,15 +83,16 @@ func (p *statusSensors) loop(name string, every time.Duration, read func(context
 			}
 			b := p.acc[x.ID]
 			if b == nil {
-				b = &bucket{start: x.T}
+				b = &bucket{}
 				p.acc[x.ID] = b
 			}
 			b.sum, b.n = b.sum+v, b.n+1
-			if x.T-b.start < stride && b.n > 1 {
+			// due a stride after the last point kept, with half a read of slack for a late timer
+			if b.last != 0 && x.T-b.last < stride-every.Milliseconds()/2 {
 				continue // still filling this point
 			}
 			h.T, h.V = append(h.T, x.T), append(h.V, b.sum/b.n)
-			*b = bucket{start: x.T + 1}
+			*b = bucket{last: x.T}
 			if len(h.T) > statusHistory {
 				h.T, h.V = h.T[len(h.T)-statusHistory:], h.V[len(h.V)-statusHistory:]
 			}

internal/metrics/fold_test.go

@@ -67,3 +67,33 @@ func TestStrideAndTrim(t *testing.T) {
 		t.Errorf("a window past the maximum is the maximum, got %v", c.Window())
 	}
 }
+
+// Points are one stride apart — not a stride and a sample — and jitter doesn't change that.
+func TestKeepSpacing(t *testing.T) {
+	c := &Collector{interval: 2 * time.Second, window: time.Hour} // stride 12 s: six samples a point
+	at := int64(1_000_000)
+	for i := 0; i < 6*40; i++ {
+		jitter := int64((i*37)%41 - 20) // ±20 ms, deterministic
+		c.keep(Sample{T: at + jitter, CPU: float64(i % 6)})
+		at += 2000
+	}
+	if len(c.ring) != 40 {
+		t.Fatalf("240 samples at six a point are 40 points, got %d", len(c.ring))
+	}
+	for i := 1; i < len(c.ring); i++ {
+		if gap := c.ring[i].T - c.ring[i-1].T; gap < 11900 || gap > 12100 {
+			t.Fatalf("point %d is %d ms after the one before; a stride is 12000", i, gap)
+		}
+	}
+	if c.ring[3].CPU != 2.5 {
+		t.Errorf("each point is the mean of its own six samples (0…5 → 2.5), got %v", c.ring[3].CPU)
+	}
+	// and when the stride is the interval, every sample is a point
+	c = &Collector{interval: 2 * time.Second, window: DefaultWindow}
+	for i := 0; i < 50; i++ {
+		c.keep(Sample{T: int64(i)*2000 + int64(i%3) - 1})
+	}
+	if len(c.ring) != 50 {
+		t.Errorf("at the default window every sample is kept, got %d of 50", len(c.ring))
+	}
+}

internal/metrics/metrics.go

@@ -32,15 +32,15 @@ const (
 	DefaultInterval = 2 * time.Second
 	MinInterval     = 100 * time.Millisecond
 	MaxInterval     = 10 * time.Second
-	History         = 300              // points kept per curve; the window below sets how far apart they are
+	History         = 300 // points kept per curve; the window below sets how far apart they are
 	// How far back the kept history reaches. The point count is fixed, so this sets the spacing:
 	// at a minute every sample is its own point, at a day each point is the average of ~5 minutes.
 	DefaultWindow = 10 * time.Minute
 	MinWindow     = 1 * time.Minute
 	MaxWindow     = 24 * time.Hour
-	fsEvery         = 30 * time.Second // filesystem usage changes slowly and statfs on network mounts isn't free
-	upsEvery        = 10 * time.Second // UPS daemons are an exec each; don't hammer them at fast intervals
-	sectorSize      = 512              // /proc/diskstats always counts 512-byte sectors
+	fsEvery       = 30 * time.Second // filesystem usage changes slowly and statfs on network mounts isn't free
+	upsEvery      = 10 * time.Second // UPS daemons are an exec each; don't hammer them at fast intervals
+	sectorSize    = 512              // /proc/diskstats always counts 512-byte sectors
 )
 
 type Mem struct {
@@ -146,6 +146,7 @@ type Collector struct {
 	ring     []Sample // the kept history: one point per stride, each the average of its bucket
 	last     Sample   // the newest raw sample, whatever the stride
 	bucket   []Sample // samples since the last point was kept
+	lastKept int64    // when that was, unix ms
 	interval time.Duration
 	window   time.Duration
 	wake     chan struct{} // nudges Run when the interval changes
@@ -271,9 +272,20 @@ func (c *Collector) stride() time.Duration {
 // two-second instant happened to land on the mark.
 func (c *Collector) keep(s Sample) {
 	c.bucket = append(c.bucket, s)
-	if s.T-c.bucket[0].T < c.stride().Milliseconds() && len(c.bucket) > 1 {
+	// A point is due a stride after the last one (for the first, a stride after the interval its
+	// first sample stands for). Half an interval of slack, because a timer never fires exactly on
+	// time: without it a tick that lands 5 ms early is "not yet", and every point runs one sample
+	// long. Measured from the last point kept, not by counting samples, so a machine that stalled
+	// or slept simply keeps the next sample it gets.
+	iv := c.interval.Milliseconds()
+	since := c.lastKept
+	if since == 0 {
+		since = c.bucket[0].T - iv
+	}
+	if s.T-since < c.stride().Milliseconds()-iv/2 {
 		return
 	}
+	c.lastKept = s.T
 	c.ring = append(c.ring, fold(c.bucket))
 	c.bucket = c.bucket[:0]
 	c.trim()

internal/store/store.go

@@ -22,6 +22,7 @@ type State struct {
 	Dashboard  json.RawMessage      `json:"dashboard,omitempty"`   // UI layout prefs, opaque to the server
 	UI         json.RawMessage      `json:"ui,omitempty"`          // theme etc., opaque to the server
 	Alerts     json.RawMessage      `json:"alerts,omitempty"`      // rules, channels and recent events (internal/api/alerts.go)
+	SamplerMs  int                  `json:"sampler_ms,omitempty"`  // how often this machine samples; the hub keeps every machine on the console's refresh rate
 	HistorySec int                  `json:"history_sec,omitempty"` // how far back the graphs reach; the hub keeps every machine on the same one
 }