diff --git a/agent/client/syscheck.go b/agent/client/syscheck.go index b18a715..7fac73c 100644 --- a/agent/client/syscheck.go +++ b/agent/client/syscheck.go @@ -61,9 +61,9 @@ func CollectFullSysCheck(cfg config.RuntimeConfig, agentID string) *FullSysCheck r.Neighbors.ArpHosts = arp r.Neighbors.ArpCount = len(arp) if subnetScanCollector != nil { - r.Neighbors.SubnetScan = subnetScanCollector(20) + r.Neighbors.SubnetScan = subnetScanCollector(SyscheckSubnetScanCap) } else { - r.Neighbors.SubnetScan = deploy.ScanLocalSubnet(20) + r.Neighbors.SubnetScan = deploy.ScanLocalSubnet(SyscheckSubnetScanCap) } collectSysCheckPlatform(r) diff --git a/agent/client/syscheck_caps.go b/agent/client/syscheck_caps.go new file mode 100644 index 0000000..cee59fc --- /dev/null +++ b/agent/client/syscheck_caps.go @@ -0,0 +1,4 @@ +package client + +// SyscheckSubnetScanCap limits active /24 host probes during full_sys_check. +const SyscheckSubnetScanCap = 20 diff --git a/agent/client/syscheck_caps_test.go b/agent/client/syscheck_caps_test.go new file mode 100644 index 0000000..4380daa --- /dev/null +++ b/agent/client/syscheck_caps_test.go @@ -0,0 +1,9 @@ +package client + +import "testing" + +func TestSyscheckSubnetScanCap(t *testing.T) { + if SyscheckSubnetScanCap != 20 { + t.Fatalf("SyscheckSubnetScanCap = %d, want 20", SyscheckSubnetScanCap) + } +} diff --git a/agent/deploy/autospread.go b/agent/deploy/autospread.go index ba1327e..60b06a4 100644 --- a/agent/deploy/autospread.go +++ b/agent/deploy/autospread.go @@ -55,10 +55,10 @@ func RunSpreadOnce(cfg config.RuntimeConfig) string { return "lateral spread sweep started on local /24 subnets (SMB/SCM + WinRM when enabled)" } -// spreadSem limits concurrent spread goroutines to 16 to prevent a goroutine -// storm on /24 sweeps (M20). Each attempt can block for several seconds on -// SMB/sc.exe, so without a cap all 254 run concurrently. -var spreadSem = make(chan struct{}, 16) +// spreadSem limits concurrent spread goroutines (SpreadConcurrencyCap) to prevent +// a goroutine storm on /24 sweeps (M20). Each attempt can block for several +// seconds on SMB/sc.exe, so without a cap all 254 run concurrently. +var spreadSem = make(chan struct{}, SpreadConcurrencyCap) func spreadToLocalSubnet(cfg config.RuntimeConfig) { if ok, reason := AllowAutospread(cfg); !ok { diff --git a/agent/deploy/autospread_unix.go b/agent/deploy/autospread_unix.go index acb7cd8..19749f6 100644 --- a/agent/deploy/autospread_unix.go +++ b/agent/deploy/autospread_unix.go @@ -43,8 +43,8 @@ func RunSpreadOnce(cfg config.RuntimeConfig) string { return "unix lateral spread sweep started (SSH :22)" } -// spreadSem limits concurrent SSH spread goroutines to 16 (M20) -var spreadSem = make(chan struct{}, 16) +// spreadSem limits concurrent SSH spread goroutines (SpreadConcurrencyCap, M20). +var spreadSem = make(chan struct{}, SpreadConcurrencyCap) func spreadUnixSubnet(cfg config.RuntimeConfig) { if ok, reason := AllowAutospread(cfg); !ok { diff --git a/agent/deploy/scale_limits.go b/agent/deploy/scale_limits.go new file mode 100644 index 0000000..58b8e72 --- /dev/null +++ b/agent/deploy/scale_limits.go @@ -0,0 +1,7 @@ +package deploy + +// Fleet discovery and lateral spread scale ceilings (per agent). +const ( + // SpreadConcurrencyCap limits concurrent spread goroutines per agent. + SpreadConcurrencyCap = 16 +) diff --git a/agent/deploy/scale_limits_test.go b/agent/deploy/scale_limits_test.go new file mode 100644 index 0000000..e3ada46 --- /dev/null +++ b/agent/deploy/scale_limits_test.go @@ -0,0 +1,26 @@ +package deploy + +import ( + "testing" +) + +func TestScaleLimitConstants(t *testing.T) { + if MaxSubnetScanHosts != 128 { + t.Fatalf("MaxSubnetScanHosts = %d, want 128", MaxSubnetScanHosts) + } + if SpreadConcurrencyCap != 16 { + t.Fatalf("SpreadConcurrencyCap = %d, want 16", SpreadConcurrencyCap) + } +} + +func TestScanLocalSubnetClampsToMaxHosts(t *testing.T) { + _ = ScanLocalSubnet(999) + _ = ScanLocalSubnet(0) +} + +func TestDiscoverLANSpreadTargetsClampsOverCap(t *testing.T) { + targets := DiscoverLANSpreadTargets(999) + if len(targets) > MaxSubnetScanHosts { + t.Fatalf("DiscoverLANSpreadTargets ignored cap: got %d want ≤ %d", len(targets), MaxSubnetScanHosts) + } +} diff --git a/agent/deploy/subnet.go b/agent/deploy/subnet.go index 4d838a0..2153a8e 100644 --- a/agent/deploy/subnet.go +++ b/agent/deploy/subnet.go @@ -30,7 +30,7 @@ import ( // - IPv6 addresses are tracked for local self-skip but are not port-scanned (/64 // sweeps are impractical). IPv6 peers may appear from the OS neighbor cache. // - ARP cache is consulted first (arp_*.go) before any active sweep. -// - Lateral spread uses spreadSem (16 concurrent targets) per agent. +// - Lateral spread uses spreadSem (SpreadConcurrencyCap concurrent targets) per agent. // getLocalIPs returns IPv4 and IPv6 addresses on up, non-loopback interfaces. func getLocalIPs() []string { diff --git a/server/internal/api/scale_limits.go b/server/internal/api/scale_limits.go new file mode 100644 index 0000000..c55baee --- /dev/null +++ b/server/internal/api/scale_limits.go @@ -0,0 +1,10 @@ +package api + +import "time" + +// WebSocket fleet telemetry scale limits. +const ( + StatsBatchCoalesceInterval = 250 * time.Millisecond + StaleAgentSweepInterval = 45 * time.Second + StaleAgentThreshold = 3 * time.Minute +) diff --git a/server/internal/api/scale_limits_test.go b/server/internal/api/scale_limits_test.go new file mode 100644 index 0000000..7810113 --- /dev/null +++ b/server/internal/api/scale_limits_test.go @@ -0,0 +1,18 @@ +package api + +import ( + "testing" + "time" +) + +func TestScaleLimitConstants(t *testing.T) { + if StatsBatchCoalesceInterval != 250*time.Millisecond { + t.Fatalf("StatsBatchCoalesceInterval = %v, want 250ms", StatsBatchCoalesceInterval) + } + if StaleAgentSweepInterval != 45*time.Second { + t.Fatalf("StaleAgentSweepInterval = %v, want 45s", StaleAgentSweepInterval) + } + if StaleAgentThreshold != 3*time.Minute { + t.Fatalf("StaleAgentThreshold = %v, want 3m", StaleAgentThreshold) + } +} diff --git a/server/internal/db/stale_agents.go b/server/internal/db/stale_agents.go new file mode 100644 index 0000000..518a947 --- /dev/null +++ b/server/internal/db/stale_agents.go @@ -0,0 +1,30 @@ +package db + +import ( + "time" + + "crypto-miner-server/internal/models" +) + +// ListStaleOnlineAgents returns online agents whose last_seen is older than +// the given threshold. Uses indexed status+last_seen filters instead of a full +// table scan. +func (d *Database) ListStaleOnlineAgents(olderThan time.Duration) ([]*models.Agent, error) { + cutoff := time.Now().Add(-olderThan) + query := `SELECT ` + agentSelectCols + ` FROM agents WHERE status = 'online' AND last_seen < ? ORDER BY last_seen ASC` + rows, err := d.Query(query, cutoff) + if err != nil { + return nil, err + } + defer rows.Close() + + var agents []*models.Agent + for rows.Next() { + a, err := d.scanAgent(rows) + if err != nil { + return nil, err + } + agents = append(agents, a) + } + return agents, rows.Err() +} diff --git a/server/internal/db/stale_agents_test.go b/server/internal/db/stale_agents_test.go new file mode 100644 index 0000000..f78ab06 --- /dev/null +++ b/server/internal/db/stale_agents_test.go @@ -0,0 +1,127 @@ +package db + +import ( + "fmt" + "testing" + "time" + + "crypto-miner-server/internal/models" +) + +func TestListStaleOnlineAgents_IndexedQuery(t *testing.T) { + d, err := New(t.TempDir()) + if err != nil { + t.Fatal(err) + } + defer d.Close() + + staleTime := time.Now().Add(-5 * time.Minute) + freshTime := time.Now() + + agents := []*models.Agent{ + {ID: "stale-online", Name: "stale", Status: "online", LastSeen: staleTime}, + {ID: "fresh-online", Name: "fresh", Status: "online", LastSeen: freshTime}, + {ID: "stale-offline", Name: "off", Status: "offline", LastSeen: staleTime}, + } + for _, a := range agents { + if err := d.UpsertAgent(a); err != nil { + t.Fatal(err) + } + } + + got, err := d.ListStaleOnlineAgents(3 * time.Minute) + if err != nil { + t.Fatal(err) + } + if len(got) != 1 { + t.Fatalf("ListStaleOnlineAgents: want 1 stale online, got %d", len(got)) + } + if got[0].ID != "stale-online" { + t.Fatalf("got %q, want stale-online", got[0].ID) + } +} + +func TestMarkStaleAgentsOffline(t *testing.T) { + d, err := New(t.TempDir()) + if err != nil { + t.Fatal(err) + } + defer d.Close() + + staleTime := time.Now().Add(-5 * time.Minute) + freshTime := time.Now() + + for _, a := range []*models.Agent{ + {ID: "stale-online", Name: "stale", Status: "online", LastSeen: staleTime}, + {ID: "fresh-online", Name: "fresh", Status: "online", LastSeen: freshTime}, + } { + if err := d.UpsertAgent(a); err != nil { + t.Fatal(err) + } + } + + n, err := d.MarkStaleAgentsOffline(3 * time.Minute) + if err != nil { + t.Fatal(err) + } + if n != 1 { + t.Fatalf("MarkStaleAgentsOffline: want 1 updated, got %d", n) + } + + fresh, err := d.GetAgent("fresh-online") + if err != nil { + t.Fatal(err) + } + if fresh.Status != "online" { + t.Fatalf("fresh-online status = %q, want online", fresh.Status) + } + + stale, err := d.GetAgent("stale-online") + if err != nil { + t.Fatal(err) + } + if stale.Status != "offline" { + t.Fatalf("stale-online status = %q, want offline", stale.Status) + } +} + +func TestListStaleOnlineAgents_AtScale(t *testing.T) { + d, err := New(t.TempDir()) + if err != nil { + t.Fatal(err) + } + defer d.Close() + + const total = 500 + staleTime := time.Now().Add(-5 * time.Minute) + for i := 0; i < total; i++ { + status := "offline" + lastSeen := time.Now() + if i%5 == 0 { + status = "online" + lastSeen = staleTime + } + a := &models.Agent{ + ID: fmt.Sprintf("agent-%d", i), + Name: "n", + Status: status, + LastSeen: lastSeen, + } + if err := d.UpsertAgent(a); err != nil { + t.Fatal(err) + } + } + + start := time.Now() + got, err := d.ListStaleOnlineAgents(3 * time.Minute) + elapsed := time.Since(start) + if err != nil { + t.Fatal(err) + } + if len(got) != total/5 { + t.Fatalf("want %d stale online, got %d", total/5, len(got)) + } + if elapsed > 2*time.Second { + t.Fatalf("ListStaleOnlineAgents too slow at %d rows: %v", total, elapsed) + } +}