fix(agent): time out blocked fan reads so collection can't hang (#2638, #2639)

Fan sensors were read with a plain os.ReadFile and no timeout. An i915 fan1_input read that blocks in the kernel held the agent lock forever, stalling every hub request and the WebSocket read loop, so the system went down and the agent never reconnected. Fan reads now go through the same timeout as temperatures (SENSORS_TIMEOUT, default 2s). A shared readWithTimeout helper tracks a read still blocked after timing out and skips later polls until it returns, instead of leaking another blocked goroutine each time. The temperature path uses the helper too, fixing the same pile-up there.
2026-10-21 22:02:34 +00:00 · 2026-10-19 13:03:13 -04:00
parent 7179bf8b8a
commit b431db751f
5 changed files with 126 additions and 19 deletions
@@ -10,6 +10,7 @@ import (
 	"net"
 	"strings"
 	"sync"
+	"sync/atomic"
 	"time"

 	"github.com/gliderlabs/ssh"
@@ -40,6 +41,8 @@ type Agent struct {
 	netInterfaceDeltaTrackers map[uint16]*deltatracker.DeltaTracker[string, uint64] // Per-cache-time NIC delta trackers
 	dockerManager             *dockerManager                                        // Manages Docker API requests
 	sensorConfig              *SensorConfig                                         // Sensors config
+	tempsReading              atomic.Bool                                           // A temperature read is still running (possibly blocked in the kernel)
+	fansReading               atomic.Bool                                           // A fan read is still running (possibly blocked in the kernel)
 	systemInfo                system.Info                                           // Host system info (dynamic)
 	systemDetails             system.Details                                        // Host system details (static, once-per-connection)
 	detailsDirty              bool                                                  // Whether system details have changed and need to be resent
@@ -1,6 +1,7 @@
 package agent

 import (
+	"errors"
 	"log/slog"
 	"os"
 	"path/filepath"
@@ -11,6 +12,8 @@ import (
 	"github.com/henrygd/beszel/internal/entities/system"
 )

+var errFanFetchTimeout = errors.New("fan collection timed out")
+
 type fanSensor struct {
 	key, path, chip string
 }
@@ -29,16 +32,15 @@ func (a *Agent) updateFans(systemStats *system.Stats) {
 	if hwmonRoot == "" {
 		return
 	}
-	sensors, err := getFanSensors()
+	fans, err := a.readFansWithTimeout(getFanSensors)
 	if err != nil {
-		slog.Debug("Error reading fans", "err", err)
+		if errors.Is(err, errFanFetchTimeout) {
+			slog.Warn("Error reading fans", "err", err)
+		} else {
+			slog.Debug("Error reading fans", "err", err)
+		}
 		return
 	}
-	// Filter before reading fan*_input: each read can wake an idle GPU.
-	if a.sensorConfig != nil && a.sensorConfig.skipGPU {
-		sensors = filterGpuFans(sensors)
-	}
-	fans := readFanSensors(sensors)
 	if len(fans) == 0 {
 		return
 	}
@@ -56,6 +58,27 @@ func (a *Agent) updateFans(systemStats *system.Stats) {
 	// }
 }

+// readFansWithTimeout discovers and reads fan sensors without letting a blocked
+// sysfs read (e.g. an i915 fan1_input stuck resuming the GPU) hang collection,
+// which holds the agent lock and would stall every later hub request.
+func (a *Agent) readFansWithTimeout(getSensors func() ([]fanSensor, error)) (map[string]uint16, error) {
+	timeout := defaultSensorsTimeout
+	if a.sensorConfig != nil && a.sensorConfig.timeout > 0 {
+		timeout = a.sensorConfig.timeout
+	}
+	return readWithTimeout(&a.fansReading, timeout, errFanFetchTimeout, func() (map[string]uint16, error) {
+		sensors, err := getSensors()
+		if err != nil {
+			return nil, err
+		}
+		// Filter before reading fan*_input: each read can wake an idle GPU.
+		if a.sensorConfig != nil && a.sensorConfig.skipGPU {
+			sensors = filterGpuFans(sensors)
+		}
+		return readFanSensors(sensors), nil
+	})
+}
+
 // readHwmonFans walks the given hwmon root (typically /sys/class/hwmon) and
 // returns a map of "<chip>_<label-or-fan-idx>" → RPM for every fan*_input
 // file it finds. Zero RPM is retained because it can represent a real fan that
@@ -6,6 +6,7 @@ import (
 	"os"
 	"path/filepath"
 	"testing"
+	"time"

 	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/require"
@@ -120,3 +121,37 @@ func TestFilterGpuFans(t *testing.T) {
 	assert.Equal(t, "nct6798_fan1", filtered[0].key)
 	assert.Len(t, discovered, 2)
 }
+
+func TestReadFansWithTimeout(t *testing.T) {
+	agent := &Agent{sensorConfig: &SensorConfig{timeout: 10 * time.Millisecond}}
+	release := make(chan struct{})
+	blocked := func() ([]fanSensor, error) {
+		<-release
+		return nil, nil
+	}
+
+	fans, err := agent.readFansWithTimeout(blocked)
+	assert.ErrorIs(t, err, errFanFetchTimeout)
+	assert.Nil(t, fans)
+
+	// a read still blocked in the kernel isn't retried on later polls
+	calls := 0
+	fans, err = agent.readFansWithTimeout(func() ([]fanSensor, error) {
+		calls++
+		return nil, nil
+	})
+	assert.ErrorIs(t, err, errSensorReadInFlight)
+	assert.Nil(t, fans)
+	assert.Zero(t, calls)
+
+	// once the blocked read returns, collection resumes
+	close(release)
+	require.Eventually(t, func() bool { return !agent.fansReading.Load() }, time.Second, time.Millisecond)
+
+	root := t.TempDir()
+	writeFile(t, filepath.Join(root, "hwmon0", "name"), "chip\n")
+	writeFile(t, filepath.Join(root, "hwmon0", "fan1_input"), "1000\n")
+	fans, err = agent.readFansWithTimeout(newFanSensorCache(root))
+	require.NoError(t, err)
+	assert.Equal(t, map[string]uint16{"chip_fan1": 1000}, fans)
+}
@@ -11,6 +11,7 @@ import (
 	"runtime"
 	"strconv"
 	"strings"
+	"sync/atomic"
 	"time"
 	"unicode/utf8"

@@ -21,7 +22,12 @@ import (
 	"github.com/shirou/gopsutil/v4/sensors"
 )

-var errTemperatureFetchTimeout = errors.New("temperature collection timed out")
+var (
+	errTemperatureFetchTimeout = errors.New("temperature collection timed out")
+	errSensorReadInFlight      = errors.New("previous sensor read still in progress")
+)
+
+const defaultSensorsTimeout = 2 * time.Second

 // Matches sensors.TemperaturesWithContext to allow for panic recovery (gopsutil/issues/1832)
 type getTempsFn func(ctx context.Context) ([]sensors.TemperatureStat, error)
@@ -53,7 +59,7 @@ func (a *Agent) newSensorConfig() *SensorConfig {
 // newSensorConfigWithEnv creates a SensorConfig with the provided environment variables
 // sensorsSet indicates if the SENSORS environment variable was explicitly set (even to empty string)
 func (a *Agent) newSensorConfigWithEnv(primarySensor, sysSensors, sensorsEnvVal, sensorsTimeout string, skipCollection, skipGPU bool) *SensorConfig {
-	timeout := 2 * time.Second
+	timeout := defaultSensorsTimeout
 	if sensorsTimeout != "" {
 		if d, err := time.ParseDuration(sensorsTimeout); err == nil {
 			timeout = d
@@ -126,7 +132,7 @@ func (a *Agent) updateTemperatures(systemStats *system.Stats) {
 	temps, err := a.getTempsWithTimeout(getSensorTemps)
 	if err != nil {
 		// retry once on panic (gopsutil/issues/1832)
-		if !errors.Is(err, errTemperatureFetchTimeout) {
+		if !errors.Is(err, errTemperatureFetchTimeout) && !errors.Is(err, errSensorReadInFlight) {
 			temps, err = a.getTempsWithTimeout(getSensorTemps)
 		}
 		if err != nil {
@@ -195,11 +201,6 @@ func (a *Agent) getTempsWithPanicRecovery(getTemps getTempsFn) (temps []sensors.
 }

 func (a *Agent) getTempsWithTimeout(getTemps getTempsFn) ([]sensors.TemperatureStat, error) {
-	type result struct {
-		temps []sensors.TemperatureStat
-		err   error
-	}
-
 	// Use a longer timeout on the first run to allow for initialization
 	// (e.g. Windows LHM subprocess startup)
 	timeout := a.sensorConfig.timeout
@@ -208,17 +209,37 @@ func (a *Agent) getTempsWithTimeout(getTemps getTempsFn) ([]sensors.TemperatureS
 		timeout = 10 * time.Second
 	}

+	return readWithTimeout(&a.tempsReading, timeout, errTemperatureFetchTimeout, func() ([]sensors.TemperatureStat, error) {
+		return a.getTempsWithPanicRecovery(getTemps)
+	})
+}
+
+// readWithTimeout runs read in a goroutine and returns timeoutErr if it doesn't
+// finish within timeout. A sysfs read blocked in the kernel can't be cancelled,
+// so inflight stays set until it returns and later calls get errSensorReadInFlight
+// instead of stacking up another blocked goroutine on every poll.
+func readWithTimeout[T any](inflight *atomic.Bool, timeout time.Duration, timeoutErr error, read func() (T, error)) (T, error) {
+	var zero T
+	if !inflight.CompareAndSwap(false, true) {
+		return zero, errSensorReadInFlight
+	}
+
+	type result struct {
+		value T
+		err   error
+	}
 	resultCh := make(chan result, 1)
 	go func() {
-		temps, err := a.getTempsWithPanicRecovery(getTemps)
-		resultCh <- result{temps: temps, err: err}
+		value, err := read()
+		inflight.Store(false)
+		resultCh <- result{value, err}
 	}()

 	select {
 	case res := <-resultCh:
-		return res.temps, res.err
+		return res.value, res.err
 	case <-time.After(timeout):
-		return nil, errTemperatureFetchTimeout
+		return zero, timeoutErr
 	}
 }

@@ -594,6 +594,31 @@ func TestGetTempsWithTimeout(t *testing.T) {
 		assert.Nil(t, temps)
 		assert.ErrorIs(t, err, errTemperatureFetchTimeout)
 	})
+
+	t.Run("skips while a timed out read is still blocked", func(t *testing.T) {
+		agent := &Agent{sensorConfig: &SensorConfig{context: context.Background(), timeout: 10 * time.Millisecond}}
+		release := make(chan struct{})
+		_, err := agent.getTempsWithTimeout(func(ctx context.Context) ([]sensors.TemperatureStat, error) {
+			<-release
+			return nil, nil
+		})
+		assert.ErrorIs(t, err, errTemperatureFetchTimeout)
+
+		called := false
+		_, err = agent.getTempsWithTimeout(func(ctx context.Context) ([]sensors.TemperatureStat, error) {
+			called = true
+			return nil, nil
+		})
+		assert.ErrorIs(t, err, errSensorReadInFlight)
+		assert.False(t, called)
+
+		close(release)
+		require.Eventually(t, func() bool { return !agent.tempsReading.Load() }, time.Second, time.Millisecond)
+		_, err = agent.getTempsWithTimeout(func(ctx context.Context) ([]sensors.TemperatureStat, error) {
+			return nil, nil
+		})
+		assert.NoError(t, err)
+	})
 }

 func TestUpdateTemperaturesSkipsOnTimeout(t *testing.T) {