Fix phantom online agents: startup reset, init reconciliation, stale-sweep goroutine
This commit is contained in:
@@ -119,7 +119,13 @@ type WSHub struct {
|
||||
}
|
||||
|
||||
func NewWSHub(database *db.Database) *WSHub {
|
||||
return &WSHub{
|
||||
// Reset any rows that were left "online" by a previous server crash/restart.
|
||||
// Agents will re-authenticate and flip themselves back to online within seconds.
|
||||
if err := database.MarkAllAgentsOffline(); err != nil {
|
||||
log.Printf("[hub] startup offline reset: %v", err)
|
||||
}
|
||||
|
||||
h := &WSHub{
|
||||
db: database,
|
||||
agents: make(map[string]*AgentConnection),
|
||||
dashboards: make(map[string]*DashboardConn),
|
||||
@@ -129,6 +135,53 @@ func NewWSHub(database *db.Database) *WSHub {
|
||||
agentDNS: make(map[string][]string),
|
||||
pingIntervalSec: 30,
|
||||
}
|
||||
|
||||
// Background stale-agent sweep: if an agent's last_seen is more than
|
||||
// 3 minutes old but the row still says "online", force it offline.
|
||||
// This catches TCP half-open drops that slip past the ping/pong timeout.
|
||||
go h.runStaleAgentSweep()
|
||||
|
||||
return h
|
||||
}
|
||||
|
||||
// runStaleAgentSweep periodically marks online agents offline when their
|
||||
// last_seen timestamp is stale (> 3 minutes without a stats message).
|
||||
// It also notifies the dashboard so client state stays in sync.
|
||||
func (h *WSHub) runStaleAgentSweep() {
|
||||
const staleness = 3 * time.Minute
|
||||
ticker := time.NewTicker(45 * time.Second)
|
||||
defer ticker.Stop()
|
||||
for range ticker.C {
|
||||
// Only sweep agents that are NOT currently connected in memory.
|
||||
// If a live WS exists, let normal disconnect handling do its job.
|
||||
h.mu.RLock()
|
||||
liveIDs := make(map[string]bool, len(h.agents))
|
||||
for id := range h.agents {
|
||||
liveIDs[id] = true
|
||||
}
|
||||
h.mu.RUnlock()
|
||||
|
||||
agents, err := h.db.ListAgents()
|
||||
if err != nil {
|
||||
continue
|
||||
}
|
||||
for _, a := range agents {
|
||||
if a == nil || a.Status != "online" || liveIDs[a.ID] {
|
||||
continue
|
||||
}
|
||||
if time.Since(a.LastSeen) < staleness {
|
||||
continue
|
||||
}
|
||||
// Row claims online, no live socket, last_seen is stale — fix it.
|
||||
_ = h.db.SetAgentOffline(a.ID)
|
||||
h.broadcastDashboard(Message{
|
||||
Type: "agent_offline",
|
||||
Payload: mustMarshal(map[string]string{"agent_id": a.ID}),
|
||||
})
|
||||
log.Printf("[hub] stale-sweep marked agent %s offline (last_seen %s ago)",
|
||||
a.ID, time.Since(a.LastSeen).Round(time.Second))
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
func (h *WSHub) SetServerPolicy(p ServerPolicy) {
|
||||
@@ -924,12 +977,19 @@ func (h *WSHub) HandleDashboardWS(w http.ResponseWriter, r *http.Request) {
|
||||
conn.Close()
|
||||
}()
|
||||
|
||||
// Send initial data
|
||||
// Send initial data — reconcile DB status against live hub state so a
|
||||
// freshly loaded dashboard never shows stale "online" phantoms.
|
||||
agents, _ := h.db.ListAgents()
|
||||
h.enrichAgentsCapabilities(agents)
|
||||
for _, a := range agents {
|
||||
if a != nil && h.isAgentConnected(a.ID) {
|
||||
if a == nil {
|
||||
continue
|
||||
}
|
||||
if h.isAgentConnected(a.ID) {
|
||||
a.Status = "online"
|
||||
} else {
|
||||
// Correct any row that says "online" but has no live socket.
|
||||
a.Status = "offline"
|
||||
}
|
||||
}
|
||||
stats, _ := h.db.GetFleetStats()
|
||||
|
||||
@@ -169,6 +169,30 @@ func (d *Database) SetAgentOffline(id string) error {
|
||||
return err
|
||||
}
|
||||
|
||||
// MarkAllAgentsOffline resets every agent row to offline. Called once at
|
||||
// server startup so rows left online by a previous crash are corrected
|
||||
// before any agent has had a chance to reconnect.
|
||||
func (d *Database) MarkAllAgentsOffline() error {
|
||||
_, err := d.Exec("UPDATE agents SET status = 'offline' WHERE status = 'online'")
|
||||
return err
|
||||
}
|
||||
|
||||
// MarkStaleAgentsOffline marks agents offline when their last_seen timestamp
|
||||
// is older than the given staleness threshold. Returns the number of rows
|
||||
// updated so the caller can emit a log line when something actually changed.
|
||||
func (d *Database) MarkStaleAgentsOffline(olderThan time.Duration) (int, error) {
|
||||
cutoff := time.Now().Add(-olderThan)
|
||||
res, err := d.Exec(
|
||||
"UPDATE agents SET status = 'offline' WHERE status = 'online' AND last_seen < ?",
|
||||
cutoff,
|
||||
)
|
||||
if err != nil {
|
||||
return 0, err
|
||||
}
|
||||
n, _ := res.RowsAffected()
|
||||
return int(n), nil
|
||||
}
|
||||
|
||||
func (d *Database) DeleteAgent(id string) error {
|
||||
_, err := d.Exec("DELETE FROM agents WHERE id = ?", id)
|
||||
return err
|
||||
|
||||
Reference in New Issue
Block a user