diff --git a/.changeset/driver-poll-return-interval.md b/.changeset/driver-poll-return-interval.md new file mode 100644 index 000000000..046067ad9 --- /dev/null +++ b/.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. diff --git a/go/internal/drivers/host.go b/go/internal/drivers/host.go index 4b8636a31..2a01cf25f 100644 --- a/go/internal/drivers/host.go +++ b/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() diff --git a/go/internal/drivers/lua.go b/go/internal/drivers/lua.go index 77316cb30..af1372d26 100644 --- a/go/internal/drivers/lua.go +++ b/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 } diff --git a/go/internal/drivers/poll_interval_test.go b/go/internal/drivers/poll_interval_test.go new file mode 100644 index 000000000..a3ff1bd7a --- /dev/null +++ b/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) +} diff --git a/go/internal/drivers/registry.go b/go/internal/drivers/registry.go index 962ef00b2..222e71ea9 100644 --- a/go/internal/drivers/registry.go +++ b/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