From c3a432101b9be352ed88e4e46b545f821929ad33 Mon Sep 17 00:00:00 2001 From: TowyTowy <85077986+TowyTowy@users.noreply.github.com> Date: Fri, 14 Aug 2026 01:04:38 +0200 Subject: [PATCH] fix(hub): re-dial SSH-pull agents when a connection silently dies (#2126) The per-system updater reused an SSH client across ticks and ran the data exchange with no deadline. If a connection went half-open (dead peer that never sends RST/FIN) or an agent accepted the session but never wrote a response, the read in fetchDataViaSSH blocked forever. Because StartUpdater calls update() synchronously on its ticker, a blocked read froze the whole per-system goroutine: the ticker's subsequent ticks were dropped, no error was returned so the system stayed "up", and the agent was never re-dialed until the hub process restarted. Bound each SSH data exchange with sshOperationTimeout via runWithTimeout: on timeout the connection is torn down (unwinding the blocked read) and a retryable error is returned, so the next tick re-dials. Also enable TCP keep-alive on dialed connections as a backstop for genuine network death. Fixes #2041 --- internal/hub/systems/ssh_timeout_test.go | 56 +++++++++++++++++ internal/hub/systems/system.go | 78 +++++++++++++++++++++++- 2 files changed, 131 insertions(+), 3 deletions(-) create mode 100644 internal/hub/systems/ssh_timeout_test.go diff --git a/internal/hub/systems/ssh_timeout_test.go b/internal/hub/systems/ssh_timeout_test.go new file mode 100644 index 00000000..2e7e1373 --- /dev/null +++ b/internal/hub/systems/ssh_timeout_test.go @@ -0,0 +1,56 @@ +//go:build testing + +package systems + +import ( + "errors" + "testing" + "testing/synctest" + "time" + + "github.com/stretchr/testify/assert" +) + +// TestRunWithTimeout covers the guard added for issue #2041: the per-system SSH +// data exchange must never block the updater indefinitely on a dead connection. +func TestRunWithTimeout(t *testing.T) { + t.Run("returns the operation result when it completes before the timeout", func(t *testing.T) { + synctest.Test(t, func(t *testing.T) { + wantErr := errors.New("boom") + onTimeoutCalled := false + + retry, err := runWithTimeout(10*time.Second, func() (bool, error) { + return true, wantErr + }, func() { onTimeoutCalled = true }) + + assert.True(t, retry, "should return the operation's retry value") + assert.Equal(t, wantErr, err, "should return the operation's error") + assert.False(t, onTimeoutCalled, "onTimeout must not fire when the op completes") + }) + }) + + t.Run("times out and tears down the connection when the op blocks", func(t *testing.T) { + synctest.Test(t, func(t *testing.T) { + // unblock simulates a half-open connection: the op is stuck reading a + // response that never arrives until the connection is torn down. + unblock := make(chan struct{}) + onTimeoutCalled := false + start := time.Now() + + retry, err := runWithTimeout(5*time.Second, func() (bool, error) { + <-unblock + return false, nil + }, func() { + onTimeoutCalled = true + close(unblock) // tearing down the connection releases the blocked read + }) + + assert.Equal(t, 5*time.Second, time.Since(start), "should return exactly at the timeout") + assert.True(t, retry, "a timeout should be retryable so the next tick re-dials") + assert.Error(t, err, "a timeout must surface an error so the system is set down") + assert.True(t, onTimeoutCalled, "onTimeout must fire so the dead connection is closed") + + synctest.Wait() // ensure the released op goroutine exits cleanly + }) + }) +} diff --git a/internal/hub/systems/system.go b/internal/hub/systems/system.go index adb54e80..c92c1884 100644 --- a/internal/hub/systems/system.go +++ b/internal/hub/systems/system.go @@ -624,10 +624,17 @@ func (sys *System) runSSHOperation(timeout time.Duration, retries int, operation continue } - retry, opErr := func() (bool, error) { + // Bound the whole operation. A half-open TCP connection (a dead peer that + // never sends RST/FIN) or a wedged agent that accepts the session but + // never writes a response would otherwise block the read forever. Because + // StartUpdater runs update() synchronously on its ticker, that stalls the + // per-system updater indefinitely with no error and no re-dial until the + // hub is restarted (issue #2041). On timeout we tear down the connection + // so the blocked read unwinds and the system is re-dialed on the next tick. + retry, opErr := runWithTimeout(sshOperationTimeout, func() (bool, error) { defer session.Close() return operation(session) - }() + }, sys.closeSSHConnection) if opErr == nil { return nil @@ -646,6 +653,43 @@ func (sys *System) runSSHOperation(timeout time.Duration, retries int, operation return fmt.Errorf("ssh operation failed") } +// sshOperationTimeout bounds a single SSH data exchange (send request, read +// response, wait for the remote command to exit). It is more generous than the +// session-creation timeout to tolerate briefly slow agents, but is kept well +// under the collection interval so a stalled connection is detected and +// re-dialed within one cycle (see issue #2041). +const sshOperationTimeout = 20 * time.Second + +// runWithTimeout runs op in a goroutine and returns its result, or, if op does +// not finish within timeout, calls onTimeout (used to tear down the connection +// so a blocked op can unwind) and returns a retryable timeout error. This +// guarantees the caller can never block indefinitely on a dead SSH connection. +func runWithTimeout(timeout time.Duration, op func() (bool, error), onTimeout func()) (retry bool, err error) { + type opResult struct { + retry bool + err error + } + // Buffered so the op goroutine never leaks even when we return on timeout. + done := make(chan opResult, 1) + go func() { + r, e := op() + done <- opResult{retry: r, err: e} + }() + + timer := time.NewTimer(timeout) + defer timer.Stop() + + select { + case res := <-done: + return res.retry, res.err + case <-timer.C: + if onTimeout != nil { + onTimeout() + } + return true, fmt.Errorf("ssh operation timed out after %s", timeout) + } +} + // createSSHClient creates a new SSH client for the system func (s *System) createSSHClient() error { if s.manager.sshConfig == nil { @@ -661,7 +705,7 @@ func (s *System) createSSHClient() error { host = net.JoinHostPort(host, s.Port) } var err error - s.client, err = ssh.Dial(network, host, s.manager.sshConfig) + s.client, err = dialSSHWithKeepAlive(network, host, s.manager.sshConfig) if err != nil { return err } @@ -670,6 +714,34 @@ func (s *System) createSSHClient() error { return nil } +// sshKeepAliveInterval is the TCP keep-alive idle interval for SSH connections +// to agents. Enabling OS-level keep-alives lets the hub eventually detect a +// dead peer on an otherwise idle connection instead of trusting it forever. +// This is a backstop for genuine network death; an application-level wedge +// (agent process hung while its kernel keeps ACKing) is caught by the +// per-operation timeout in runSSHOperation instead (see issue #2041). +const sshKeepAliveInterval = 30 * time.Second + +// dialSSHWithKeepAlive dials an SSH connection like ssh.Dial, but enables TCP +// keep-alive on the underlying connection so half-open connections are +// eventually detected by the operating system. +func dialSSHWithKeepAlive(network, addr string, config *ssh.ClientConfig) (*ssh.Client, error) { + dialer := net.Dialer{ + Timeout: config.Timeout, + KeepAlive: sshKeepAliveInterval, + } + conn, err := dialer.Dial(network, addr) + if err != nil { + return nil, err + } + sshConn, chans, reqs, err := ssh.NewClientConn(conn, addr, config) + if err != nil { + _ = conn.Close() + return nil, err + } + return ssh.NewClient(sshConn, chans, reqs), nil +} + // createSessionWithTimeout creates a new SSH session with a timeout to avoid hanging // in case of network issues func (sys *System) createSessionWithTimeout(timeout time.Duration) (*ssh.Session, error) {