diff --git a/internal/hub/systems/system.go b/internal/hub/systems/system.go index 57d0f1995..7ef9e3223 100644 --- a/internal/hub/systems/system.go +++ b/internal/hub/systems/system.go @@ -174,24 +174,11 @@ func (sys *System) update() error { data, err := sys.fetchDataFromAgent(options) if err != nil { - // Keep the WebSocket alive even though this fetch failed. - // - // Only a message from the agent resets the read deadline, and an agent - // still collecting sends nothing. Without this ping the deadline expires - // between polls, the hub closes the connection, the agent reconnects, - // collects slowly again, and the cycle repeats. Pinging here is the same - // move handlePaused already makes for a paused system, and Ping() pushes - // the deadline out as it sends, so it works without an answer, which - // matters because a collecting agent cannot answer. - // - // This is why the deadline has to exceed interval + wsDataRequestTimeout: - // the ping only happens once the fetch gives up. See ws.deadline. - // - // The system is still reported down: that comes from the error we - // return. Only the connection survives, so recovery costs no reconnect. - // A failed ping needs no handling either, since a connection that - // cannot be pinged runs into its deadline anyway. - if sys.WsConn != nil && sys.WsConn.IsConnected() { + // Ping so a slow agent's connection survives until the next poll. Ping() + // extends the read deadline without needing a reply (see ws.deadline). + // Once the agent has been silent for maxAgentSilence, stop pinging and let + // the deadline close what is likely a dead connection. + if sys.WsConn != nil && sys.WsConn.IsConnected() && sys.WsConn.TimeSinceLastMessage() < maxAgentSilence { _ = sys.WsConn.Ping() } return err @@ -820,6 +807,12 @@ func (sys *System) fetchDataFromAgent(options common.DataRequestOptions) (*syste // so this must be well above the request manager's 5s default. var wsDataRequestTimeout = 30 * time.Second +// maxAgentSilence is how long a WebSocket agent may go without sending anything +// before failed polls stop keeping its connection alive. An agent whose +// collection runs past its own 120s deadline drops the connection itself, so a +// live agent always answers within interval + 120s. +var maxAgentSilence = 3 * time.Minute + func (sys *System) fetchDataViaWebSocket(options common.DataRequestOptions) (*system.CombinedData, error) { if sys.WsConn == nil || !sys.WsConn.IsConnected() { return nil, errors.New("no websocket connection") diff --git a/internal/hub/systems/ws_data_timeout_test.go b/internal/hub/systems/ws_data_timeout_test.go index a8a019c47..c50cd4632 100644 --- a/internal/hub/systems/ws_data_timeout_test.go +++ b/internal/hub/systems/ws_data_timeout_test.go @@ -185,18 +185,26 @@ const timelineScale = 100 // happens once the fetch gives up, which is interval + wsDataRequestTimeout into // the cycle, so the read deadline has to outlast that. With a deadline below // that sum the connection is already closed when the ping would go out, which is -// the reconnect loop from #2294. +// the reconnect loop from #2294. An agent silent for longer than +// maxAgentSilence is no longer pinged, so its connection is allowed to expire. func TestSlowAgentKeepsConnectionAcrossPolls(t *testing.T) { tick := time.Duration(interval) * time.Millisecond / timelineScale tests := []struct { name string deadline time.Duration + polls int wantConnected bool }{ - {"production deadline outlasts a failed poll", ws.Deadline() / timelineScale, true}, + // The first failed poll shows the ping lands, the second shows the ping + // keeps re-arming the deadline. + {"production deadline outlasts a failed poll", ws.Deadline() / timelineScale, 2, true}, // 70s was the old value and sits below the 90s a failed poll needs. - {"deadline below interval plus request timeout", 70 * time.Second / timelineScale, false}, + {"deadline below interval plus request timeout", 70 * time.Second / timelineScale, 2, false}, + // The third failed poll ends 210s after the last message, past + // maxAgentSilence, so it skips the ping and the deadline closes the + // connection before the fifth poll. + {"agent silent past maxAgentSilence", ws.Deadline() / timelineScale, 5, false}, } for _, tc := range tests { @@ -206,6 +214,9 @@ func TestSlowAgentKeepsConnectionAcrossPolls(t *testing.T) { previousTimeout := wsDataRequestTimeout wsDataRequestTimeout = 30 * time.Second / timelineScale t.Cleanup(func() { wsDataRequestTimeout = previousTimeout }) + previousSilence := maxAgentSilence + maxAgentSilence = 3 * time.Minute / timelineScale + t.Cleanup(func() { maxAgentSilence = previousSilence }) connections := make(chan *ws.WsConn, 1) // The real handler: OnClose has to run, or a dropped connection @@ -244,9 +255,7 @@ func TestSlowAgentKeepsConnectionAcrossPolls(t *testing.T) { require.NoError(t, err, "the first poll must succeed") pollZero := time.Now() - // Two failed polls: the first shows the ping lands, the second shows - // the ping keeps re-arming the deadline. - for poll := 1; poll <= 2; poll++ { + for poll := 1; poll <= tc.polls; poll++ { time.Sleep(time.Until(pollZero.Add(time.Duration(poll) * tick))) if !sys.WsConn.IsConnected() { break diff --git a/internal/hub/ws/ws.go b/internal/hub/ws/ws.go index 6b268a3ba..c040c8190 100644 --- a/internal/hub/ws/ws.go +++ b/internal/hub/ws/ws.go @@ -36,6 +36,7 @@ type WsConn struct { requestManager *RequestManager DownChan chan struct{} agentVersion semver.Version + lastMessage atomic.Int64 // unix nanoseconds of the last message from the agent } // FingerprintRecord is fingerprints collection record data in the hub @@ -66,6 +67,7 @@ func NewWsConnection(conn *gws.Conn, agentVersion semver.Version) *WsConn { agentVersion: agentVersion, } ws.conn.Store(conn) + ws.lastMessage.Store(time.Now().UnixNano()) return ws } @@ -85,6 +87,7 @@ func (h *Handler) OnMessage(conn *gws.Conn, message *gws.Message) { _ = conn.WriteClose(1000, nil) return } + wsConn.(*WsConn).lastMessage.Store(time.Now().UnixNano()) wsConn.(*WsConn).requestManager.handleResponse(message) } @@ -126,6 +129,11 @@ func (ws *WsConn) Ping() error { return conn.WritePing(nil) } +// TimeSinceLastMessage returns how long it has been since the agent last sent a message. +func (ws *WsConn) TimeSinceLastMessage() time.Duration { + return time.Since(time.Unix(0, ws.lastMessage.Load())) +} + // sendMessage encodes data to CBOR and sends it as a binary message to the agent. // This is kept for backwards compatibility but new actions should use RequestManager. func (ws *WsConn) sendMessage(data common.HubRequest[any]) error {