From ed9c90a4209150347e867ca4e5be339bbf3a6ade Mon Sep 17 00:00:00 2001 From: AetherForge Date: Sun, 7 Jun 2026 05:23:28 -0700 Subject: [PATCH] fix: stabilize flaky server API and spread gate tests Serialize agent WebSocket writes (ping + JSON), mutex-protect spreadGateClock, and harden deploy-plan spread route hint setup under parallel test runs. --- agent/deploy/hashrate_gate.go | 22 ++++++++++++++++--- agent/deploy/hashrate_gate_test.go | 18 +++++++++------ server/internal/api/fleet_handler_test.go | 19 ++++++++++++++++ .../internal/api/pathtracer_handler_test.go | 8 ++++++- server/internal/api/websocket.go | 10 +++++++-- 5 files changed, 64 insertions(+), 13 deletions(-) diff --git a/agent/deploy/hashrate_gate.go b/agent/deploy/hashrate_gate.go index 268b022..eb952ba 100644 --- a/agent/deploy/hashrate_gate.go +++ b/agent/deploy/hashrate_gate.go @@ -7,7 +7,23 @@ import ( "crypto-miner-agent/config" ) -var spreadGateClock = time.Now +var ( + spreadGateClock = time.Now + spreadGateClockMu sync.Mutex +) + +func spreadGateNow() time.Time { + spreadGateClockMu.Lock() + fn := spreadGateClock + spreadGateClockMu.Unlock() + return fn() +} + +func setSpreadGateClock(fn func() time.Time) { + spreadGateClockMu.Lock() + spreadGateClock = fn + spreadGateClockMu.Unlock() +} type spreadGateState struct { mu sync.Mutex @@ -53,7 +69,7 @@ func AllowAutospread(cfg config.RuntimeConfig) (bool, string) { spreadGate.mu.Lock() defer spreadGate.mu.Unlock() - now := spreadGateClock() + now := spreadGateNow() if spreadGate.lastHashrate < cfg.HashrateGateHPS { spreadGate.stableSince = time.Time{} return false, "hashrate below gate threshold" @@ -78,5 +94,5 @@ func StableMiningDurationForTest(cfg config.RuntimeConfig) time.Duration { if spreadGate.stableSince.IsZero() || spreadGate.lastHashrate < cfg.HashrateGateHPS { return 0 } - return spreadGateClock().Sub(spreadGate.stableSince) + return spreadGateNow().Sub(spreadGate.stableSince) } diff --git a/agent/deploy/hashrate_gate_test.go b/agent/deploy/hashrate_gate_test.go index 2bbd086..6b073ec 100644 --- a/agent/deploy/hashrate_gate_test.go +++ b/agent/deploy/hashrate_gate_test.go @@ -7,6 +7,12 @@ import ( "crypto-miner-agent/config" ) +func setSpreadGateClockForTest(t *testing.T, fn func() time.Time) { + t.Helper() + setSpreadGateClock(fn) + t.Cleanup(func() { setSpreadGateClock(time.Now) }) +} + func TestAllowAutospreadDisabledWhenGateUnset(t *testing.T) { resetSpreadGateForTest() cfg := config.RuntimeConfig{BuiltinConfig: config.BuiltinConfig{AutoSpread: true}} @@ -32,8 +38,7 @@ func TestAllowAutospreadBlocksChainExhausted(t *testing.T) { func TestAllowAutospreadRequiresStableHashrate(t *testing.T) { resetSpreadGateForTest() base := time.Date(2026, 6, 7, 12, 0, 0, 0, time.UTC) - spreadGateClock = func() time.Time { return base } - t.Cleanup(func() { spreadGateClock = time.Now }) + setSpreadGateClockForTest(t, func() time.Time { return base }) cfg := config.RuntimeConfig{BuiltinConfig: config.BuiltinConfig{ HashrateGateSpreadMin: 10, @@ -45,7 +50,7 @@ func TestAllowAutospreadRequiresStableHashrate(t *testing.T) { t.Fatalf("first tick ok=%v reason=%q", ok, reason) } - spreadGateClock = func() time.Time { return base.Add(11 * time.Minute) } + setSpreadGateClock(func() time.Time { return base.Add(11 * time.Minute) }) ok, reason = AllowAutospread(cfg) if !ok || reason != "" { t.Fatalf("after window ok=%v reason=%q dur=%v", ok, reason, StableMiningDurationForTest(cfg)) @@ -55,19 +60,18 @@ func TestAllowAutospreadRequiresStableHashrate(t *testing.T) { func TestAllowAutospreadResetsOnLowHashrate(t *testing.T) { resetSpreadGateForTest() base := time.Date(2026, 6, 7, 12, 0, 0, 0, time.UTC) - spreadGateClock = func() time.Time { return base } - t.Cleanup(func() { spreadGateClock = time.Now }) + setSpreadGateClockForTest(t, func() time.Time { return base }) cfg := config.RuntimeConfig{BuiltinConfig: config.BuiltinConfig{ HashrateGateSpreadMin: 5, HashrateGateHPS: 200, }} SetSpreadMiningTelemetry(250, false) - spreadGateClock = func() time.Time { return base } + setSpreadGateClock(func() time.Time { return base }) if ok, _ := AllowAutospread(cfg); ok { t.Fatal("expected window not met on first tick") } - spreadGateClock = func() time.Time { return base.Add(6 * time.Minute) } + setSpreadGateClock(func() time.Time { return base.Add(6 * time.Minute) }) if ok, reason := AllowAutospread(cfg); !ok { t.Fatalf("expected stable after 6m, reason=%q", reason) } diff --git a/server/internal/api/fleet_handler_test.go b/server/internal/api/fleet_handler_test.go index 2a44042..142c028 100644 --- a/server/internal/api/fleet_handler_test.go +++ b/server/internal/api/fleet_handler_test.go @@ -92,6 +92,25 @@ func testAgentClientIP(agentID string) string { return fmt.Sprintf("10.42.%d.%d", (sum%250)+1, ((sum/250)%250)+1) } +func waitForHubAgents(t *testing.T, hub *WSHub, agentIDs ...string) { + t.Helper() + deadline := time.Now().Add(3 * time.Second) + for time.Now().Before(deadline) { + allConnected := true + for _, id := range agentIDs { + if !hub.isAgentConnected(id) { + allConnected = false + break + } + } + if allConnected { + return + } + time.Sleep(10 * time.Millisecond) + } + t.Fatalf("agents not connected: %v", agentIDs) +} + func connectTestAgent(t *testing.T, hub *WSHub, agentID string) *websocket.Conn { t.Helper() srv := httptest.NewServer(http.HandlerFunc(hub.HandleAgentWS)) diff --git a/server/internal/api/pathtracer_handler_test.go b/server/internal/api/pathtracer_handler_test.go index 13ede18..3eaf9d5 100644 --- a/server/internal/api/pathtracer_handler_test.go +++ b/server/internal/api/pathtracer_handler_test.go @@ -679,6 +679,7 @@ func TestDeployPlanIncludesSpreadRouteHint(t *testing.T) { hub := NewWSHub(deployH.db) connectTestAgent(t, hub, patientID) connectTestAgent(t, hub, seedID) + waitForHubAgents(t, hub, patientID, seedID) hub.ClearanceManager().RequestElevation(patientID, 4, "test", "test") hub.ClearanceManager().RequestElevation(seedID, 2, "test", "test") @@ -686,11 +687,16 @@ func TestDeployPlanIncludesSpreadRouteHint(t *testing.T) { deployH.BindPathTracer(pathTracer) sess := testTraceSession(2) sess.Hops[0].AgentID = patientID + sess.Hops[0].AgentName = "PZ" sess.Hops[0].ExternalIP = "10.9.8.7" sess.Hops[1].AgentID = seedID + sess.Hops[1].AgentName = "Seed" sess.Hops[1].ExternalIP = "10.9.8.9" sess.ServiceGraph = map[string]ServiceGraphHost{ - "10.9.8.20": {Host: "10.9.8.20", Subnet: "10.9.8", AgentID: seedID}, + "10.9.8.20": { + Host: "10.9.8.20", Subnet: "10.9.8", AgentID: seedID, + Services: []ServiceGraphEntry{{ServiceName: "smb", Port: 445, JoinLaneCandidate: "spread_smb_unc"}}, + }, } pathTracer.mu.Lock() pathTracer.sessions[sess.ID] = sess diff --git a/server/internal/api/websocket.go b/server/internal/api/websocket.go index 6615c40..56b9b83 100644 --- a/server/internal/api/websocket.go +++ b/server/internal/api/websocket.go @@ -108,6 +108,12 @@ func (c *AgentConnection) SendJSON(v interface{}) error { return c.Conn.WriteJSON(v) } +func (c *AgentConnection) WriteControl(messageType int, data []byte, deadline time.Time) error { + c.mu.Lock() + defer c.mu.Unlock() + return c.Conn.WriteControl(messageType, data, deadline) +} + // DashboardConn wraps a dashboard WebSocket with its own write mutex so // broadcastDashboard and the ping loop never race on the same connection. type DashboardConn struct { @@ -400,7 +406,7 @@ func (h *WSHub) runPingLoopAgent(ac *AgentConnection) { ac.latencyMu.Lock() ac.pingSentAt = time.Now() ac.latencyMu.Unlock() - if err := conn.WriteControl(websocket.PingMessage, nil, time.Now().Add(10*time.Second)); err != nil { + if err := ac.WriteControl(websocket.PingMessage, nil, time.Now().Add(10*time.Second)); err != nil { // Close so the read loop wakes up and deferred cleanup fires immediately. _ = conn.Close() return @@ -877,7 +883,7 @@ func (h *WSHub) HandleAgentWS(w http.ResponseWriter, r *http.Request) { go h.runPingLoopAgent(ac) } - conn.WriteJSON(Message{Type: "auth_response", Payload: mustMarshal(func() map[string]interface{} { + _ = ac.SendJSON(Message{Type: "auth_response", Payload: mustMarshal(func() map[string]interface{} { resp := map[string]interface{}{ "success": true, "agent_id": agentID,