Merge pull request #1990 from rcourtman/maintainer/20260908T165532Z

Keep TrueNAS sessions alive and refresh on a steady cadence

docs/release-control/v6/internal/subsystems/monitoring.md

@@ -17,6 +17,33 @@
 
 ## Purpose
 
+### TrueNAS persistent-session liveness and successful poll cadence
+
+Authenticated JSON-RPC WebSocket sessions send transport-only PING controls
+every 25 seconds with a five-second write deadline. The sender belongs to the
+session, not its opening request context, and disposal stops and joins it.
+A failed control write closes the socket so existing transport handling owns
+retry; actions retain their no-replay boundary. Existing serialized RPC and
+stream readers consume pongs. Neither ping nor pong updates inventory freshness
+or appliance health, and sending keepalives is not proactive pong-timeout detection.
+
+Successful TrueNAS refreshes target start-to-start cadence, bounded below by
+completion plus min(five seconds, configured interval), preventing back-to-back
+load on slow appliances. Failed refreshes retain a full completion-based retry
+interval. Manual connection tests retain completion-based scheduling. Observed
+last-attempt and last-success timestamps remain completion timestamps; resource
+freshness thresholds must not be extended to hide genuinely stale devices.
+
+`TestAuthenticatedRPCSurvivesIdleTransportTimeout` and
+`TestRPCSessionKeepaliveConcurrentCallsAndShutdown` in
+`internal/truenas/transport_test.go` verify an authenticated synthetic idle-timeout
+server, opening-context cancellation, concurrent controls/RPCs and sender disposal.
+`TestTrueNASSuccessfulPollCadenceIncludesBoundedIdleGap` in
+`internal/monitoring/truenas_poller_test.go` verifies due boundaries, short/slow
+cycles, completion timestamps and unchanged failure backoff. These are synthetic
+runtime proofs, not native firmware timeout or reporter-resolution evidence.
+
+
 **Availability backfill preserves concurrent discovery changes (7 September 2026)**
 
 The backfill List snapshot is a work list, not an authoritative record to save.

internal/monitoring/truenas_poller.go

@@ -517,7 +517,7 @@ func (p *TrueNASPoller) pollAll(ctx context.Context) {
 
 		snapshot := entry.provider.Snapshot()
 		p.mu.Lock()
-		p.recordConnectionSuccessLocked(entry.orgID, entry.id, entry.config, end, snapshot)
+		p.recordConnectionSuccessLocked(entry.orgID, entry.id, entry.config, start, end, snapshot)
 		p.mu.Unlock()
 		refreshedOrgs[entry.orgID] = struct{}{}
 		p.ingestRecoveryPoints(ctx, entry.orgID, entry.id, entry.provider)
@@ -811,6 +811,7 @@ func (p *TrueNASPoller) recordConnectionSuccessLocked(
 	orgID string,
 	connID string,
 	instance config.TrueNASInstance,
+	startedAt time.Time,
 	at time.Time,
 	snapshot *truenas.FixtureSnapshot,
 ) {
@@ -819,7 +820,16 @@ func (p *TrueNASPoller) recordConnectionSuccessLocked(
 	status.lastSuccessAt = at
 	status.lastError = nil
 	status.consecutiveFailures = 0
-	status.nextPollAt = at.Add(p.effectiveRuntimePollInterval(instance))
+	// Successful cycles target start-to-start cadence, but never immediately
+	// hammer a slow appliance with another refresh. Keep at least five seconds
+	// idle (or the entire interval for short-interval configurations). Failures
+	// continue to use a full completion-based retry interval below.
+	interval := p.effectiveRuntimePollInterval(instance)
+	idleGap := min(5*time.Second, interval)
+	status.nextPollAt = startedAt.Add(interval)
+	if earliest := at.Add(idleGap); status.nextPollAt.Before(earliest) {
+		status.nextPollAt = earliest
+	}
 	if snapshot != nil {
 		status.observed = buildTrueNASObservedSummary(snapshot)
 	}
@@ -869,7 +879,7 @@ func (p *TrueNASPoller) RecordConnectionTestSuccess(
 
 	p.mu.Lock()
 	defer p.mu.Unlock()
-	p.recordConnectionSuccessLocked(orgID, connID, instance, at, nil)
+	p.recordConnectionSuccessLocked(orgID, connID, instance, at, at, nil)
 }
 
 // RecordConnectionTestFailure updates one saved TrueNAS connection summary after

internal/monitoring/truenas_poller_test.go

@@ -414,7 +414,7 @@ func TestTrueNASPollerManualConnectionTestsUpdateSummariesWithoutClearingObserve
 	manualSuccessAt := failureAt.Add(2 * time.Minute)
 
 	poller.mu.Lock()
-	poller.recordConnectionSuccessLocked("default", connection.ID, connection, firstSuccess, snapshot)
+	poller.recordConnectionSuccessLocked("default", connection.ID, connection, firstSuccess, firstSuccess, snapshot)
 	poller.recordConnectionFailureLocked("default", connection.ID, connection, errors.New("manual auth failed"), failureAt)
 	poller.mu.Unlock()
 
@@ -2164,3 +2164,44 @@ func TestTrueNASPollerKeysSystemsByConnection(t *testing.T) {
 		return firstOK && secondOK && len(ids) == 2
 	}, "expected one connection-scoped system source ID per configured connection")
 }
+
+func TestTrueNASSuccessfulPollCadenceIncludesBoundedIdleGap(t *testing.T) {
+	for _, tc := range []struct {
+		name                     string
+		interval, duration, next time.Duration
+	}{
+		{"fast", time.Minute, 2 * time.Second, time.Minute},
+		{"nearly_due", time.Minute, 58 * time.Second, 63 * time.Second},
+		{"slow", time.Minute, 90 * time.Second, 95 * time.Second},
+		{"genuinely_stale", time.Minute, 122 * time.Second, 127 * time.Second},
+		{"short_interval", time.Second, 2 * time.Second, 3 * time.Second},
+	} {
+		t.Run(tc.name, func(t *testing.T) {
+			poller := NewTrueNASPoller(nil, 0, nil)
+			instance := config.TrueNASInstance{ID: "cadence", PollIntervalSecs: int(tc.interval / time.Second)}
+			start := time.Date(2026, 9, 8, 0, 0, 0, 0, time.UTC)
+			end := start.Add(tc.duration)
+			poller.recordConnectionSuccessLocked("default", instance.ID, instance, start, end, nil)
+			status := poller.ensureConnectionRuntimeStatusLocked("default", instance.ID)
+			want := start.Add(tc.next)
+			if !status.nextPollAt.Equal(want) {
+				t.Errorf("next poll = %v, want %v", status.nextPollAt, want)
+			}
+			if !status.lastSuccessAt.Equal(end) || !status.lastAttemptAt.Equal(end) {
+				t.Error("scheduling altered observed completion timestamps")
+			}
+			if poller.connectionPollDueLocked("default", instance.ID, instance, want.Add(-time.Nanosecond)) {
+				t.Error("poll due before bounded idle gap elapsed")
+			}
+			if !poller.connectionPollDueLocked("default", instance.ID, instance, want) {
+				t.Error("poll not due at scheduled time")
+			}
+			// Failed refreshes retain the full interval after completion; this change
+			// must not accelerate retries against an unavailable appliance.
+			poller.recordConnectionFailureLocked("default", instance.ID, instance, errors.New("offline"), end)
+			if !status.nextPollAt.Equal(end.Add(tc.interval)) {
+				t.Error("failure backoff changed")
+			}
+		})
+	}
+}

internal/truenas/client.go

@@ -2061,8 +2061,10 @@ func appendDiskTemperature(out map[string]int, diskName string, value any) {
 }
 
 type trueNASRPCClient struct {
-	conn   *websocket.Conn
-	nextID int64
+	conn          *websocket.Conn
+	nextID        int64
+	keepaliveStop chan struct{}
+	keepaliveDone chan struct{}
 }
 
 func (c *trueNASRPCClient) subscribe(ctx context.Context, event string) (string, error) {

internal/truenas/transport.go

@@ -9,6 +9,8 @@ import (
 	"strconv"
 	"strings"
 	"time"
+
+	"github.com/gorilla/websocket"
 )
 
 var errRPCStreamSessionConsumed = errors.New("truenas rpc stream session cannot be reused")
@@ -276,6 +278,7 @@ func (c *Client) openAuthenticatedRPC(ctx context.Context) (*trueNASRPCClient, s
 		_ = conn.Close()
 		return nil, "", err
 	}
+	rpc.startKeepalive(25 * time.Second)
 	return rpc, authMechanism, nil
 }
 
@@ -444,7 +447,7 @@ func (c *Client) waitReconnectBackoff(ctx context.Context) error {
 
 func (c *Client) closeRPCLocked() {
 	if c.rpc != nil && c.rpc.conn != nil {
-		_ = c.rpc.conn.Close()
+		c.rpc.close()
 	}
 	c.rpc = nil
 	c.updateTransportStatus(func(status *TransportStatus) {
@@ -452,6 +455,45 @@ func (c *Client) closeRPCLocked() {
 	})
 }
 
+// startKeepalive belongs to the authenticated session, not the context of the
+// call which opened it. WriteControl is safe alongside the serialized RPC
+// reader/writer; it must not change their deadlines or acquire rpcMu.
+// Pongs are consumed by the existing RPC/stream readers. A ping is only idle
+// transport maintenance, never evidence of fresh inventory or appliance health.
+func (c *trueNASRPCClient) startKeepalive(interval time.Duration) {
+	c.keepaliveStop = make(chan struct{})
+	c.keepaliveDone = make(chan struct{})
+	go func() {
+		defer close(c.keepaliveDone)
+		ticker := time.NewTicker(interval)
+		defer ticker.Stop()
+		for {
+			select {
+			case <-c.keepaliveStop:
+				return
+			case <-ticker.C:
+				if err := c.conn.WriteControl(websocket.PingMessage, nil, time.Now().Add(5*time.Second)); err != nil {
+					// Unblock any in-flight reader; ordinary transport handling owns retry
+					// and status, including the no-replay rule for actions.
+					_ = c.conn.Close()
+					return
+				}
+			}
+		}
+	}()
+}
+
+// The owning Client serializes session disposal with rpcMu.
+func (c *trueNASRPCClient) close() {
+	if c.keepaliveStop != nil {
+		close(c.keepaliveStop)
+	}
+	_ = c.conn.Close()
+	if c.keepaliveDone != nil {
+		<-c.keepaliveDone
+	}
+}
+
 func (c *Client) recordTransportError(err error) {
 	message := c.sanitizeTransportError(err)
 	c.updateTransportStatus(func(status *TransportStatus) {

internal/truenas/transport_test.go

@@ -993,3 +993,105 @@ func TestIssue1631HTTPSUpgradeTargetRules(t *testing.T) {
 		}
 	}
 }
+
+// Model an authenticated appliance with a transport idle limit. This is not
+// evidence of a particular appliance firmware's authenticated timeout policy.
+func TestAuthenticatedRPCSurvivesIdleTransportTimeout(t *testing.T) {
+	var sessions, pings atomic.Int32
+	upgrader := websocket.Upgrader{}
+	server := httptest.NewTLSServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		conn, err := upgrader.Upgrade(w, r, nil)
+		if err != nil {
+			return
+		}
+		defer conn.Close()
+		sessions.Add(1)
+		resetIdle := func() { _ = conn.SetReadDeadline(time.Now().Add(35 * time.Second)) }
+		conn.SetPingHandler(func(data string) error {
+			pings.Add(1)
+			resetIdle()
+			return conn.WriteControl(websocket.PongMessage, []byte(data), time.Now().Add(time.Second))
+		})
+		resetIdle()
+		for {
+			var request trueNASRPCRequest
+			if err := conn.ReadJSON(&request); err != nil {
+				return
+			}
+			resetIdle()
+			var result any = map[string]any{"version": "TrueNAS-SCALE-25.04.2"}
+			if request.Method == "auth.login_ex" {
+				result = map[string]any{"response_type": "SUCCESS"}
+			}
+			if err := conn.WriteJSON(map[string]any{"jsonrpc": "2.0", "id": request.ID, "result": result}); err != nil {
+				return
+			}
+		}
+	}))
+	defer server.Close()
+	client := protocolFixtureClient(t, server.URL, ClientConfig{APIKey: "fixture-key", Username: "fixture-user"})
+	defer client.Close()
+	ctx, cancel := context.WithTimeout(context.Background(), 50*time.Second)
+	defer cancel()
+	if _, err := client.GetSystemInfo(ctx); err != nil {
+		t.Fatal(err)
+	}
+	client.rpcMu.Lock()
+	session := client.rpc
+	client.rpcMu.Unlock()
+	// Cancelling the opening call must not cancel the persistent session.
+	cancel()
+	ctx, cancel = context.WithTimeout(context.Background(), 50*time.Second)
+	defer cancel()
+	// Longer than the fixture's idle timeout, shorter than two production pings.
+	time.Sleep(40 * time.Second)
+	if _, err := client.GetSystemInfo(ctx); err != nil {
+		t.Fatal(err)
+	}
+	if got := sessions.Load(); got != 1 {
+		t.Errorf("authenticated sessions = %d, want 1 (idle connection was lost)", got)
+	}
+	if pings.Load() == 0 {
+		t.Error("no websocket keepalive received")
+	}
+	client.Close()
+	select {
+	case <-session.keepaliveDone:
+	default:
+		t.Error("Close returned before keepalive exited")
+	}
+}
+
+func TestRPCSessionKeepaliveConcurrentCallsAndShutdown(t *testing.T) {
+	fixture := newProtocolFixture(t, func(_ int, _ trueNASRPCRequest) protocolFixtureReply {
+		time.Sleep(3 * time.Millisecond)
+		return protocolFixtureReply{result: "ok"}
+	}, nil)
+	client := protocolFixtureClient(t, fixture.server.URL, ClientConfig{})
+	ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
+	defer cancel()
+	conn, err := client.dialRPC(ctx)
+	if err != nil {
+		t.Fatal(err)
+	}
+	rpc := &trueNASRPCClient{conn: conn, nextID: 1}
+	rpc.startKeepalive(time.Millisecond)
+	defer rpc.close()
+	for i := 0; i < 20; i++ {
+		var result string
+		if err := rpc.call(ctx, "fixture.read", nil, &result); err != nil {
+			t.Fatal(err)
+		}
+		if result != "ok" {
+			t.Fatalf("result = %q", result)
+		}
+	}
+	// A failed control write must terminate its sender and close the socket,
+	// rather than retaining a goroutine until a future poll/client shutdown.
+	_ = conn.Close()
+	select {
+	case <-rpc.keepaliveDone:
+	case <-time.After(time.Second):
+		t.Fatal("keepalive did not exit after socket failure")
+	}
+}