fix: stabilize flaky server API and spread gate tests
Some checks failed
CI Docker Mining Proof / Linux agent hashrate proof (push) Has been cancelled
Some checks failed
CI Docker Mining Proof / Linux agent hashrate proof (push) Has been cancelled
Serialize agent WebSocket writes (ping + JSON), mutex-protect spreadGateClock, and harden deploy-plan spread route hint setup under parallel test runs.
This commit is contained in:
@@ -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)
|
||||
}
|
||||
|
||||
@@ -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)
|
||||
}
|
||||
|
||||
@@ -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))
|
||||
|
||||
@@ -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
|
||||
|
||||
@@ -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,
|
||||
|
||||
Reference in New Issue
Block a user