feat: speed test through the browser, client - panel - browser (0.0.6)

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

internal/server/handlers_panel_stat.go

@@ -2,6 +2,7 @@ package server
 
 import (
 	"crypto/rand"
+	"encoding/json"
 	"errors"
 	"io"
 	"net/http"
@@ -70,8 +71,8 @@ func (s *Server) handleStatusPage(w http.ResponseWriter, r *http.Request) {
 }
 
 // runTask runs a one-shot task on the agent and waits for its result.
-func (s *Server) runTask(agentID, typ string, timeout time.Duration) (proto.TaskResult, error) {
-	taskID, ch := s.hub.CreateTask(agentID, typ, nil)
+func (s *Server) runTask(agentID, typ string, args []byte, timeout time.Duration) (proto.TaskResult, error) {
+	taskID, ch := s.hub.CreateTask(agentID, typ, args)
 	defer s.hub.DiscardTask(taskID)
 	t := time.NewTimer(timeout)
 	defer t.Stop()
@@ -118,7 +119,7 @@ func (s *Server) handleAgentStat(w http.ResponseWriter, r *http.Request) {
 	e.mu.Lock()
 	defer e.mu.Unlock()
 	if time.Since(e.at) > statCacheFor {
-		res, err := s.runTask(id, proto.TaskSysStat, statTaskTimeout)
+		res, err := s.runTask(id, proto.TaskSysStat, nil, statTaskTimeout)
 		if err != nil {
 			// Cached too, so pages queued behind this one don't each wait it out.
 			res = proto.TaskResult{Error: err.Error()}
@@ -129,17 +130,39 @@ func (s *Server) handleAgentStat(w http.ResponseWriter, r *http.Request) {
 }
 
 // POST /api/panel/agents/{id}/speed - have the agent measure its link to us.
+// With relay=1 its streams run through the relay opened for it, to and from
+// the browser.
 func (s *Server) handleAgentSpeed(w http.ResponseWriter, r *http.Request) {
 	id, ok := s.onlineAgent(w, r)
 	if !ok {
 		return
 	}
-	res, err := s.runTask(id, proto.TaskSysSpeed, speedTaskTimeout)
+	relay := r.URL.Query().Get("relay") != ""
+	var args []byte
+	if relay {
+		rl := s.relays.get(id)
+		if rl == nil {
+			writeJSON(w, proto.TaskResult{Error: "замер через браузер не открыт"})
+			return
+		}
+		defer s.relays.drop(id, rl)
+		args, _ = json.Marshal(proto.SysSpeedArgs{Relay: true})
+	}
+	res, err := s.runTask(id, proto.TaskSysSpeed, args, speedTaskTimeout)
 	if err != nil {
 		writeJSON(w, proto.TaskResult{Error: err.Error()})
 		return
 	}
-	writeJSON(w, oldAgentError(res))
+	res = oldAgentError(res)
+	if relay && res.OK {
+		// An agent that predates the relay ignores the argument and measures
+		// its own link.
+		var d proto.SysSpeedResult
+		if json.Unmarshal(res.Data, &d) != nil || !d.Relay {
+			res = proto.TaskResult{Error: "клиент не умеет замер через браузер - обновите его (нужен 0.0.6+)"}
+		}
+	}
+	writeJSON(w, res)
 }
 
 // speedBlock is one chunk of incompressible bytes, made once and sent over and
@@ -150,6 +173,13 @@ var speedBlock = sync.OnceValue(func() []byte {
 	return b
 })
 
+func speedStreamHeaders(w http.ResponseWriter) {
+	w.Header().Set("Content-Type", "application/octet-stream")
+	w.Header().Set("Cache-Control", "no-store")
+	// Keep a proxy in front from buffering or compressing the stream.
+	w.Header().Set("X-Accel-Buffering", "no")
+}
+
 // 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) {
@@ -163,10 +193,7 @@ func (s *Server) handleSpeedDown(w http.ResponseWriter, r *http.Request) {
 			left = n
 		}
 	}
-	w.Header().Set("Content-Type", "application/octet-stream")
-	w.Header().Set("Cache-Control", "no-store")
-	// Keep a proxy in front from buffering or compressing the stream.
-	w.Header().Set("X-Accel-Buffering", "no")
+	speedStreamHeaders(w)
 	block := speedBlock()
 	deadline := time.Now().Add(speedDownFor)
 	for left > 0 && time.Now().Before(deadline) && r.Context().Err() == nil {
@@ -191,3 +218,148 @@ func (s *Server) handleSpeedUp(w http.ResponseWriter, r *http.Request) {
 	_, _ = io.Copy(io.Discard, http.MaxBytesReader(w, r.Body, speedUpCap))
 	w.WriteHeader(http.StatusNoContent)
 }
+
+// speedRelay joins the agent's speed test streams to the browser's, so the
+// bytes travel browser - panel - client. Both pipes are unbuffered: a slow
+// reader holds its writer back, and the rate is that of the whole path.
+type speedRelay struct {
+	downR *io.PipeReader // browser -> agent
+	downW *io.PipeWriter
+	upR   *io.PipeReader // agent -> browser
+	upW   *io.PipeWriter
+}
+
+func (rl *speedRelay) close() {
+	rl.downR.CloseWithError(io.ErrClosedPipe)
+	rl.downW.CloseWithError(io.ErrClosedPipe)
+	rl.upR.CloseWithError(io.ErrClosedPipe)
+	rl.upW.CloseWithError(io.ErrClosedPipe)
+}
+
+// speedRelays holds the relay of each agent under test, one per agent.
+type speedRelays struct {
+	mu sync.Mutex
+	m  map[string]*speedRelay
+}
+
+// open starts a fresh relay for the agent, ending the one it had.
+func (c *speedRelays) open(agentID string) {
+	rl := &speedRelay{}
+	rl.downR, rl.downW = io.Pipe()
+	rl.upR, rl.upW = io.Pipe()
+	c.mu.Lock()
+	if c.m == nil {
+		c.m = map[string]*speedRelay{}
+	}
+	old := c.m[agentID]
+	c.m[agentID] = rl
+	c.mu.Unlock()
+	if old != nil {
+		old.close()
+	}
+}
+
+func (c *speedRelays) get(agentID string) *speedRelay {
+	c.mu.Lock()
+	defer c.mu.Unlock()
+	return c.m[agentID]
+}
+
+// drop ends the agent's relay; with rl given, only if it is still that one.
+func (c *speedRelays) drop(agentID string, rl *speedRelay) {
+	c.mu.Lock()
+	cur := c.m[agentID]
+	if cur != nil && (rl == nil || rl == cur) {
+		delete(c.m, agentID)
+	} else {
+		cur = nil
+	}
+	c.mu.Unlock()
+	if cur != nil {
+		cur.close()
+	}
+}
+
+// GET /api/panel/ping - an empty answer for the browser to time.
+func (s *Server) handlePanelPing(w http.ResponseWriter, r *http.Request) {
+	w.Header().Set("Cache-Control", "no-store")
+	w.WriteHeader(http.StatusNoContent)
+}
+
+// POST /api/panel/agents/{id}/speed/relay - open the relay of a speed test
+// that runs through the browser. The test itself ends it.
+func (s *Server) handleSpeedRelayOpen(w http.ResponseWriter, r *http.Request) {
+	id, ok := s.onlineAgent(w, r)
+	if !ok {
+		return
+	}
+	s.relays.open(id)
+	writeJSON(w, proto.TaskResult{OK: true})
+}
+
+// POST /api/panel/agents/{id}/speed/down - a piece of the browser's upload,
+// passed on to the agent. 410 once the agent has stopped reading.
+func (s *Server) handleSpeedRelayFeed(w http.ResponseWriter, r *http.Request) {
+	rl := s.relays.get(r.PathValue("id"))
+	if rl == nil {
+		http.Error(w, "gone", http.StatusGone)
+		return
+	}
+	if _, err := io.Copy(rl.downW, http.MaxBytesReader(w, r.Body, speedUpCap)); err != nil {
+		http.Error(w, "gone", http.StatusGone)
+		return
+	}
+	w.WriteHeader(http.StatusNoContent)
+}
+
+// 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"))
+	if rl == nil {
+		http.Error(w, "gone", http.StatusGone)
+		return
+	}
+	speedStreamHeaders(w)
+	_, _ = io.Copy(w, rl.upR)
+	// The browser may have left: don't let the agent keep sending to nobody.
+	rl.upR.CloseWithError(io.ErrClosedPipe)
+}
+
+// GET /api/agent/speed/relay/down - what the browser uploads, for the agent to
+// time.
+func (s *Server) handleSpeedRelayDown(w http.ResponseWriter, r *http.Request) {
+	a := s.authenticateAgent(r)
+	if a == nil {
+		http.Error(w, "unauthorized", http.StatusUnauthorized)
+		return
+	}
+	rl := s.relays.get(a.ID)
+	if rl == nil {
+		http.NotFound(w, r)
+		return
+	}
+	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)
+	// The agent hung up at the end of its window: stop the browser's upload.
+	rl.downR.CloseWithError(io.ErrClosedPipe)
+}
+
+// POST /api/agent/speed/relay/up - the agent's upload, passed on to the browser.
+func (s *Server) handleSpeedRelayUp(w http.ResponseWriter, r *http.Request) {
+	a := s.authenticateAgent(r)
+	if a == nil {
+		http.Error(w, "unauthorized", http.StatusUnauthorized)
+		return
+	}
+	rl := s.relays.get(a.ID)
+	if rl == nil {
+		http.NotFound(w, r)
+		return
+	}
+	_, _ = io.Copy(rl.upW, http.MaxBytesReader(w, r.Body, speedUpCap))
+	rl.upW.Close()
+	w.WriteHeader(http.StatusNoContent)
+}