feat: live reading during the speed test, and a 15 s limit that keeps what was measured

internal/server/handlers_panel_stat.go

@@ -20,8 +20,9 @@ const (
 	// statCacheFor is how long one sample answers every open status page, so
 	// several tabs cost the agent a single reading.
 	statCacheFor = 2 * time.Second
-	// speedTaskTimeout covers the agent's ping + down + up windows with slack.
-	speedTaskTimeout = 30 * time.Second
+	// speedTaskTimeout is how long a speed test may take. One that is not done
+	// by then is answered with what the server has counted itself.
+	speedTaskTimeout = 15 * time.Second
 	// speedDownCap bounds one download stream; speedDownFor ends it if the
 	// agent does not hang up first.
 	speedDownCap = 256 << 20
@@ -141,6 +142,8 @@ func (s *Server) handleAgentSpeed(w http.ResponseWriter, r *http.Request) {
 		return
 	}
 	relay := r.URL.Query().Get("relay") != ""
+	m := s.meters.start(id)
+	defer s.meters.stop(id, m)
 	var args []byte
 	if relay {
 		rl := s.relays.get(id)
@@ -153,8 +156,14 @@ func (s *Server) handleAgentSpeed(w http.ResponseWriter, r *http.Request) {
 	}
 	res, err := s.runTask(id, proto.TaskSysSpeed, args, speedTaskTimeout)
 	if err != nil {
-		writeJSON(w, proto.TaskResult{Error: err.Error()})
-		return
+		part, ok := m.result()
+		if !ok {
+			writeJSON(w, proto.TaskResult{Error: err.Error()})
+			return
+		}
+		part.Relay = relay
+		data, _ := json.Marshal(part)
+		res = proto.TaskResult{OK: true, Data: data}
 	}
 	res = oldAgentError(res)
 	if relay && res.OK {
@@ -168,6 +177,148 @@ func (s *Server) handleAgentSpeed(w http.ResponseWriter, r *http.Request) {
 	writeJSON(w, res)
 }
 
+// speedMeter counts what flows through the streams of one speed test as the
+// server sees it: for the live reading on the page, and for the result of a
+// test that ran out of time.
+type speedMeter struct {
+	mu     sync.Mutex
+	start  time.Time
+	phase  string // "ping", "down", "up"
+	pingAt time.Time
+	pingMs float64
+	down   speedCount
+	up     speedCount
+}
+
+// speedCount is the bytes of one direction and when they went through. The
+// first piece only starts the clock.
+type speedCount struct {
+	n, skip     int64
+	first, last time.Time
+}
+
+func (c *speedCount) add(n int) {
+	now := time.Now()
+	if c.first.IsZero() {
+		c.first, c.skip = now, int64(n)
+	}
+	c.n += int64(n)
+	c.last = now
+}
+
+func (c *speedCount) bps() float64 {
+	d := c.last.Sub(c.first).Seconds()
+	if d <= 0 {
+		return 0
+	}
+	return float64(c.n-c.skip) * 8 / d
+}
+
+// ping notes one of the agent's ping probes: they come one right after
+// another, so the gap between two is its round trip.
+func (m *speedMeter) ping() {
+	if m == nil {
+		return
+	}
+	m.mu.Lock()
+	now := time.Now()
+	if !m.pingAt.IsZero() {
+		if ms := float64(now.Sub(m.pingAt).Microseconds()) / 1000; m.pingMs == 0 || ms < m.pingMs {
+			m.pingMs = ms
+		}
+	}
+	m.pingAt = now
+	m.mu.Unlock()
+}
+
+func (m *speedMeter) add(up bool, n int) {
+	if m == nil || n <= 0 {
+		return
+	}
+	m.mu.Lock()
+	if up {
+		m.phase = "up"
+		m.up.add(n)
+	} else {
+		m.phase = "down"
+		m.down.add(n)
+	}
+	m.mu.Unlock()
+}
+
+// result is the test as counted so far; false with nothing measured yet.
+func (m *speedMeter) result() (proto.SysSpeedResult, bool) {
+	m.mu.Lock()
+	defer m.mu.Unlock()
+	res := proto.SysSpeedResult{PingMs: m.pingMs, DownBps: m.down.bps(), UpBps: m.up.bps(), Partial: true}
+	return res, res.DownBps > 0
+}
+
+// meterWriter counts what it passes on; a nil meter counts nothing.
+type meterWriter struct {
+	w  io.Writer
+	m  *speedMeter
+	up bool
+}
+
+func (c meterWriter) Write(p []byte) (int, error) {
+	n, err := c.w.Write(p)
+	c.m.add(c.up, n)
+	return n, err
+}
+
+// speedMeters holds the meter of each agent under test.
+type speedMeters struct {
+	mu sync.Mutex
+	m  map[string]*speedMeter
+}
+
+func (c *speedMeters) start(agentID string) *speedMeter {
+	m := &speedMeter{start: time.Now(), phase: "ping"}
+	c.mu.Lock()
+	if c.m == nil {
+		c.m = map[string]*speedMeter{}
+	}
+	c.m[agentID] = m
+	c.mu.Unlock()
+	return m
+}
+
+func (c *speedMeters) get(agentID string) *speedMeter {
+	c.mu.Lock()
+	defer c.mu.Unlock()
+	return c.m[agentID]
+}
+
+// stop forgets the agent's meter if it is still m.
+func (c *speedMeters) stop(agentID string, m *speedMeter) {
+	c.mu.Lock()
+	if c.m[agentID] == m {
+		delete(c.m, agentID)
+	}
+	c.mu.Unlock()
+}
+
+// GET /api/panel/agents/{id}/speed/progress - the bytes counted so far in the
+// agent's running speed test, for the page to turn into a live rate. An empty
+// phase means no test is running.
+func (s *Server) handleSpeedProgress(w http.ResponseWriter, r *http.Request) {
+	var out struct {
+		Phase string `json:"phase"`
+		Down  int64  `json:"down"`
+		Up    int64  `json:"up"`
+		Ms    int64  `json:"ms"` // since the test began, by the server's clock
+	}
+	if m := s.meters.get(r.PathValue("id")); m != nil {
+		m.mu.Lock()
+		out.Phase, out.Down, out.Up = m.phase, m.down.n, m.up.n
+		out.Ms = time.Since(m.start).Milliseconds()
+		m.mu.Unlock()
+	}
+	w.Header().Set("Cache-Control", "no-store")
+	writeJSON(w, out)
+}
+
 // speedBlock is one chunk of incompressible bytes, made once and sent over and
 // over, so a speed test costs the server no CPU beyond the copy.
 var speedBlock = sync.OnceValue(func() []byte {
@@ -186,16 +337,21 @@ func speedStreamHeaders(w http.ResponseWriter) {
 // GET /api/agent/speed/down?bytes=N - stream N bytes (default: as much as fits
 // in the time window) for the agent to time. bytes=0 is the ping probe.
 func (s *Server) handleSpeedDown(w http.ResponseWriter, r *http.Request) {
-	if s.authenticateAgent(r) == nil {
+	a := s.authenticateAgent(r)
+	if a == nil {
 		http.Error(w, "unauthorized", http.StatusUnauthorized)
 		return
 	}
+	m := s.meters.get(a.ID)
 	left := speedDownCap
 	if v := r.URL.Query().Get("bytes"); v != "" {
 		if n, err := strconv.Atoi(v); err == nil && n >= 0 && n < left {
 			left = n
 		}
 	}
+	if left == 0 {
+		m.ping()
+	}
 	speedStreamHeaders(w)
 	block := speedBlock()
 	deadline := time.Now().Add(speedDownFor)
@@ -208,17 +364,20 @@ func (s *Server) handleSpeedDown(w http.ResponseWriter, r *http.Request) {
 		if err != nil {
 			return
 		}
+		m.add(false, n)
 		left -= n
 	}
 }
 
 // POST /api/agent/speed/up - swallow the agent's upload.
 func (s *Server) handleSpeedUp(w http.ResponseWriter, r *http.Request) {
-	if s.authenticateAgent(r) == nil {
+	a := s.authenticateAgent(r)
+	if a == nil {
 		http.Error(w, "unauthorized", http.StatusUnauthorized)
 		return
 	}
-	_, _ = io.Copy(io.Discard, http.MaxBytesReader(w, r.Body, speedUpCap))
+	sink := meterWriter{w: io.Discard, m: s.meters.get(a.ID), up: true}
+	_, _ = io.Copy(sink, http.MaxBytesReader(w, r.Body, speedUpCap))
 	w.WriteHeader(http.StatusNoContent)
 }
 
@@ -323,13 +482,14 @@ func (s *Server) handleSpeedRelayFeed(w http.ResponseWriter, r *http.Request) {
 // GET /api/panel/agents/{id}/speed/up - the agent's upload, for the browser to
 // time. Ends when the agent is done sending.
 func (s *Server) handleSpeedRelayDrain(w http.ResponseWriter, r *http.Request) {
-	rl := s.relays.get(r.PathValue("id"))
+	id := r.PathValue("id")
+	rl := s.relays.get(id)
 	if rl == nil {
 		http.Error(w, "gone", http.StatusGone)
 		return
 	}
 	speedStreamHeaders(w)
-	_, _ = io.Copy(w, rl.upR)
+	_, _ = io.Copy(meterWriter{w: w, m: s.meters.get(id), up: true}, rl.upR)
 	// The browser may have left: don't let the agent keep sending to nobody.
 	rl.upR.CloseWithError(io.ErrClosedPipe)
 }
@@ -350,7 +510,7 @@ func (s *Server) handleSpeedRelayDown(w http.ResponseWriter, r *http.Request) {
 	speedStreamHeaders(w)
 	// The agent waits on the headers, the browser may not have sent a byte yet.
 	_ = http.NewResponseController(w).Flush()
-	_, _ = io.Copy(w, rl.downR)
+	_, _ = io.Copy(meterWriter{w: w, m: s.meters.get(a.ID)}, rl.downR)
 	// The agent hung up at the end of its window: stop the browser's upload.
 	rl.downR.CloseWithError(io.ErrClosedPipe)
 }