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

@@ -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
+		})
+	})
+}

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) {