fix(drivers): honour the interval driver_poll returns

.changeset/driver-poll-return-interval.md

@@ -0,0 +1,7 @@
+---
+"ftw": patch
+---
+
+Honour the interval a driver returns from `driver_poll`, as the driver contract says. Core ignored it before and polled every five seconds. The returned wait now sets the next poll, between one second and thirty seconds. A driver that calls `host.set_poll_interval` keeps that interval, as before, whatever it returns.
+
+Bundled drivers that return one second now read once a second: ambibox_v2x, ferroamp, ferroamp_dc2_v2x, pixii_pv, sdm630, sonnen and zap. myuplink reads every thirty seconds instead of every five. easee_cloud, tesla_wall_connector and zaptec_cloud wait ten seconds after a failed read, and solaredge_legacy thirty seconds after a failed SunSpec probe.

go/internal/drivers/host.go

@@ -145,9 +145,13 @@ type HostEnv struct {
 	BatteryTelemetryOnly bool
 
 	mu sync.Mutex
-	// Desired poll interval — driver can set via host.set_poll_interval OR
-	// return it from driver_poll. We persist the last hint here.
-	PollIntervalMS int32
+	// Standing poll interval, 5 s until the driver calls
+	// host.set_poll_interval. Once it has, that interval wins over anything
+	// driver_poll returns.
+	PollIntervalMS  int32
+	pollIntervalSet bool
+	// scheduledPoll is the wait the run loop armed last, standing or returned.
+	scheduledPoll time.Duration
 	// Identity set by driver / capability layer.
 	// Make + SN are reported via host.set_make / host.set_sn.
 	// Endpoint is the protocol+host+port string set by the registry when
@@ -546,6 +550,7 @@ func (h *HostEnv) log(level int32, msg string) {
 func (h *HostEnv) setPollInterval(ms int32) {
 	h.mu.Lock()
 	h.PollIntervalMS = ms
+	h.pollIntervalSet = true
 	h.mu.Unlock()
 }
 
@@ -798,16 +803,66 @@ func (h *HostEnv) stampIdentity(raw []byte) []byte {
 	return out
 }
 
-// PollInterval returns the driver's current requested poll cadence.
+// PollInterval returns the driver's standing poll cadence.
 func (h *HostEnv) PollInterval() time.Duration {
 	h.mu.Lock()
 	defer h.mu.Unlock()
+	return h.standingPollIntervalLocked()
+}
+
+func (h *HostEnv) standingPollIntervalLocked() time.Duration {
 	if h.PollIntervalMS <= 0 {
 		return 5 * time.Second
 	}
 	return time.Duration(h.PollIntervalMS) * time.Millisecond
 }
 
+// Bounds for an interval a driver returns from driver_poll. Lane 0 frames
+// leave once a second and the control loop ticks every two seconds by
+// default, so a faster read reaches nothing but the device's bus; some
+// catalog drivers return 200 ms or 50 ms. The ceiling is half the default
+// 60 s watchdog window, so a returned wait cannot make a driver stale by
+// itself: myuplink returns 60 s after a good read. A driver that needs longer
+// sets it with host.set_poll_interval, as the vehicle drivers do.
+const (
+	minReturnedPollInterval = time.Second
+	maxReturnedPollInterval = 30 * time.Second
+)
+
+// returnedPollInterval converts milliseconds returned by driver_poll. It caps
+// before converting: a Lua number can exceed the range of time.Duration.
+func returnedPollInterval(ms float64) time.Duration {
+	ms = math.Min(ms, float64(maxReturnedPollInterval/time.Millisecond))
+	return time.Duration(ms) * time.Millisecond
+}
+
+// nextPollInterval picks the wait before the next driver_poll. An interval
+// set with host.set_poll_interval wins, unbounded, as it always has. A driver
+// that never set one gets the positive interval its last poll returned,
+// within bounds, or the 5 s default.
+func (h *HostEnv) nextPollInterval(returned time.Duration) time.Duration {
+	h.mu.Lock()
+	defer h.mu.Unlock()
+	next := h.standingPollIntervalLocked()
+	if !h.pollIntervalSet && returned > 0 {
+		next = min(max(returned, minReturnedPollInterval), maxReturnedPollInterval)
+	}
+	h.scheduledPoll = next
+	return next
+}
+
+// ScheduledPollInterval returns the wait the run loop armed last, or the
+// standing interval before the first poll has run.
+func (h *HostEnv) ScheduledPollInterval() time.Duration {
+	h.mu.Lock()
+	scheduled := h.scheduledPoll
+	h.mu.Unlock()
+	if scheduled > 0 {
+		return scheduled
+	}
+	return h.PollInterval()
+}
+
 // Identity returns (make, serial) set by the driver.
 func (h *HostEnv) Identity() (make, sn string) {
 	h.mu.Lock()

go/internal/drivers/lua.go

@@ -281,7 +281,7 @@ func (d *LuaDriver) pollLocked(ctx context.Context) (time.Duration, error) {
 	}
 	// Driver may return an int number of milliseconds.
 	if n, ok := ret.(lua.LNumber); ok && n > 0 {
-		return time.Duration(n) * time.Millisecond, nil
+		return returnedPollInterval(float64(n)), nil
 	}
 	return 0, nil
 }

go/internal/drivers/poll_interval_test.go

@@ -0,0 +1,142 @@
+package drivers
+
+import (
+	"context"
+	"testing"
+	"time"
+
+	"github.com/srcfl/ftw/go/internal/config"
+	"github.com/srcfl/ftw/go/internal/telemetry"
+)
+
+func TestNextPollIntervalHonoursTheReturnUnlessSetPollIntervalWasCalled(t *testing.T) {
+	cases := []struct {
+		name       string
+		standingMS int32 // 0: the driver never called host.set_poll_interval
+		returned   time.Duration
+		want       time.Duration
+	}{
+		{"nothing returned uses the default", 0, 0, 5 * time.Second},
+		{"a returned wait sets the next poll", 0, time.Second, time.Second},
+		{"a fast return is held to one second", 0, 200 * time.Millisecond, time.Second},
+		{"a long return is held to thirty seconds", 0, time.Minute, 30 * time.Second},
+		{"set_poll_interval wins over a shorter return", 2000, time.Second, 2 * time.Second},
+		{"set_poll_interval wins over a longer return", 2000, time.Minute, 2 * time.Second},
+		{"set_poll_interval is not bounded", 10, 0, 10 * time.Millisecond},
+	}
+	for _, tc := range cases {
+		t.Run(tc.name, func(t *testing.T) {
+			env := NewHostEnv("probe", nil)
+			if tc.standingMS > 0 {
+				env.setPollInterval(tc.standingMS)
+			}
+			if got := env.nextPollInterval(tc.returned); got != tc.want {
+				t.Fatalf("next wait = %v, want %v", got, tc.want)
+			}
+			if got := env.ScheduledPollInterval(); got != tc.want {
+				t.Fatalf("scheduled wait = %v, want %v", got, tc.want)
+			}
+		})
+	}
+}
+
+func TestReturnedPollIntervalCapsNumbersBeyondDuration(t *testing.T) {
+	for _, ms := range []float64{1e16, 1e300} {
+		if got := returnedPollInterval(ms); got != 30*time.Second {
+			t.Fatalf("returnedPollInterval(%g) = %v, want 30s", ms, got)
+		}
+	}
+}
+
+// returnedRuntime returns a fixed interval from every poll and reports when
+// each poll ran.
+type returnedRuntime struct {
+	env      *HostEnv
+	returned time.Duration
+	polls    chan time.Time
+}
+
+func (r *returnedRuntime) Init(context.Context, []byte) error { return nil }
+func (r *returnedRuntime) Poll(context.Context) (time.Duration, error) {
+	r.polls <- time.Now()
+	return r.returned, nil
+}
+func (r *returnedRuntime) Command(context.Context, []byte) error { return nil }
+func (r *returnedRuntime) DefaultMode(context.Context) error     { return nil }
+func (r *returnedRuntime) Cleanup(context.Context) error         { return nil }
+func (r *returnedRuntime) Discard()                              {}
+func (r *returnedRuntime) Env() *HostEnv                         { return r.env }
+
+// A driver that only returns its interval, as sungrow.lua does, is polled at
+// that interval rather than the 5 s default.
+func TestRunLoopWaitsTheIntervalThePollReturned(t *testing.T) {
+	tel := telemetry.NewStore()
+	r := NewRegistry(tel)
+	env := NewHostEnv("d1", tel)
+	// Bring the first poll forward without marking the interval as set by
+	// the driver; the default would otherwise delay it five seconds.
+	env.PollIntervalMS = 10
+	rt := &returnedRuntime{env: env, returned: time.Second, polls: make(chan time.Time, 8)}
+	rd := &runningDriver{
+		driver:    rt,
+		env:       env,
+		cfg:       config.Driver{Name: "d1"},
+		cmdCh:     make(chan driverCmd, 1),
+		defaultCh: make(chan driverCmd, 1),
+		stop:      make(chan bool, 1),
+		done:      make(chan struct{}),
+	}
+	r.rec["d1"] = rd
+	go r.runLoop(rd)
+	t.Cleanup(func() { r.remove("d1", true) })
+
+	next := func() time.Time {
+		t.Helper()
+		select {
+		case at := <-rt.polls:
+			return at
+		case <-time.After(3 * time.Second):
+			t.Fatal("no poll within 3 s")
+			return time.Time{}
+		}
+	}
+	first := next()
+	second := next()
+	if gap := second.Sub(first); gap < 950*time.Millisecond || gap > 1500*time.Millisecond {
+		t.Fatalf("second poll %v after the first, want the returned second", gap)
+	}
+	if got, _ := r.PollInterval("d1"); got != time.Second {
+		t.Fatalf("PollInterval = %v, want the returned 1s", got)
+	}
+}
+
+// A driver that set its own cadence keeps it, whatever driver_poll returns.
+func TestRunLoopKeepsSetPollIntervalOverTheReturn(t *testing.T) {
+	path := writeTestDriver(t, `
+local polls = 0
+function driver_init(config) host.set_poll_interval(10) end
+function driver_poll()
+  polls = polls + 1
+  host.emit_metric("polls", polls)
+  return 60000
+end
+function driver_command() end
+function driver_default_mode() end
+`)
+	tel := telemetry.NewStore()
+	reg := NewRegistry(tel)
+	if err := reg.Add(context.Background(), config.Driver{Name: "standing", Lua: path}); err != nil {
+		t.Fatal(err)
+	}
+	t.Cleanup(func() { reg.remove("standing", true) })
+
+	deadline := time.Now().Add(2 * time.Second)
+	for time.Now().Before(deadline) {
+		if value, _, _ := tel.LatestMetric("standing", "polls"); value >= 3 {
+			return
+		}
+		time.Sleep(5 * time.Millisecond)
+	}
+	value, _, _ := tel.LatestMetric("standing", "polls")
+	t.Fatalf("polled %v times in 2 s at a 10 ms set_poll_interval", value)
+}

go/internal/drivers/registry.go

@@ -1055,7 +1055,7 @@ func (r *Registry) runLoop(rd *runningDriver) {
 			// Register the poll as the active Lua call so SendDefault can
 			// cancel it the same way it cancels an in-flight command.
 			pollCtx, finishPoll := rd.beginCommand(ctx)
-			_, err := rd.driver.Poll(pollCtx)
+			returned, err := rd.driver.Poll(pollCtx)
 			finishPoll()
 			if err != nil {
 				slog.Warn("driver poll failed", "name", rd.cfg.Name, "err", err)
@@ -1073,8 +1073,8 @@ func (r *Registry) runLoop(rd *runningDriver) {
 				// re-stamps LastSuccess every tick from cached values.
 				r.tel.RecordDriverTick(rd.cfg.Name)
 			}
-			// Re-arm timer at driver's requested interval
-			interval = rd.env.PollInterval()
+			// Re-arm at the interval the poll returned, or the standing one.
+			interval = rd.env.nextPollInterval(returned)
 			timer.Reset(interval)
 		case <-recoveryC:
 			if err := attemptDefault(); err != nil {
@@ -1468,7 +1468,7 @@ func replacementContext(ctx context.Context) context.Context {
 	return context.WithoutCancel(ctx)
 }
 
-// PollInterval returns the currently requested cadence for a running driver.
+// PollInterval returns the cadence a running driver is polled at now.
 // Repository activation uses it to wait for two fresh cycles before declaring
 // a new artifact healthy.
 func (r *Registry) PollInterval(name string) (time.Duration, bool) {
@@ -1478,7 +1478,7 @@ func (r *Registry) PollInterval(name string) (time.Duration, bool) {
 	if !ok {
 		return 0, false
 	}
-	return rd.env.PollInterval(), true
+	return rd.env.ScheduledPollInterval(), true
 }
 
 // tcpAllowedHostsFor builds the effective allowlist for a TCP-capable