diff --git a/tariff/amber.go b/tariff/amber.go index 1df6588f1..69f05b61f 100644 --- a/tariff/amber.go +++ b/tariff/amber.go @@ -85,7 +85,9 @@ func (t *Amber) run(done chan error) { if err := backoff.Retry(func() error { return backoffPermanentError(t.GetJSON(t.uri, &res)) }, bo()); err != nil { - once.Do(func() { done <- err }) + if reportError(&once, done, err) { + return + } t.log.ERROR.Println(err) continue diff --git a/tariff/awattar.go b/tariff/awattar.go index 9327d8d19..325c6c529 100644 --- a/tariff/awattar.go +++ b/tariff/awattar.go @@ -70,7 +70,9 @@ func (t *Awattar) run(done chan error) { if err := backoff.Retry(func() error { return backoffPermanentError(client.GetJSON(uri, &res)) }, bo()); err != nil { - once.Do(func() { done <- err }) + if reportError(&once, done, err) { + return + } t.log.ERROR.Println(err) continue diff --git a/tariff/edf-tempo.go b/tariff/edf-tempo.go index 01270e3f0..1aed76368 100644 --- a/tariff/edf-tempo.go +++ b/tariff/edf-tempo.go @@ -124,7 +124,9 @@ func (t *EdfTempo) run(done chan error) { if err := backoff.Retry(func() error { return backoffPermanentError(t.GetJSON(uri, &res)) }, bo()); err != nil { - once.Do(func() { done <- err }) + if reportError(&once, done, err) { + return + } t.log.ERROR.Println(err) continue diff --git a/tariff/electricitymaps.go b/tariff/electricitymaps.go index fdf7ac9a0..d12c32ffc 100644 --- a/tariff/electricitymaps.go +++ b/tariff/electricitymaps.go @@ -88,7 +88,9 @@ func (t *ElectricityMaps) run(done chan error) { err = errors.New(res.Error) } - once.Do(func() { done <- err }) + if reportError(&once, done, err) { + return + } t.log.ERROR.Println(err) continue diff --git a/tariff/elering.go b/tariff/elering.go index f32d57ca3..887a1fed2 100644 --- a/tariff/elering.go +++ b/tariff/elering.go @@ -72,7 +72,9 @@ func (t *Elering) run(done chan error) { if err := backoff.Retry(func() error { return backoffPermanentError(client.GetJSON(uri, &res)) }, bo()); err != nil { - once.Do(func() { done <- err }) + if reportError(&once, done, err) { + return + } t.log.ERROR.Println(err) continue diff --git a/tariff/entsoe.go b/tariff/entsoe.go index a6152f733..03927e8d9 100644 --- a/tariff/entsoe.go +++ b/tariff/entsoe.go @@ -125,14 +125,18 @@ func (t *Entsoe) run(done chan error) { return backoff.Permanent(errors.New("invalid document name: " + doc.XMLName.Local)) } }, bo()); err != nil { - once.Do(func() { done <- err }) + if reportError(&once, done, err) { + return + } t.log.ERROR.Println(err) continue } if len(tr.TimeSeries) == 0 { - once.Do(func() { done <- entsoe.ErrInvalidData }) + if reportError(&once, done, entsoe.ErrInvalidData) { + return + } t.log.ERROR.Println(entsoe.ErrInvalidData) continue } @@ -140,7 +144,9 @@ func (t *Entsoe) run(done chan error) { // extract desired series res, err := entsoe.GetTsPriceData(tr.TimeSeries, entsoe.ResolutionQuarterHour) if err != nil { - once.Do(func() { done <- err }) + if reportError(&once, done, err) { + return + } t.log.ERROR.Println(err) continue } diff --git a/tariff/gruenstromindex.go b/tariff/gruenstromindex.go index e35541dfd..d47968f03 100644 --- a/tariff/gruenstromindex.go +++ b/tariff/gruenstromindex.go @@ -75,7 +75,9 @@ func (t *GrünStromIndex) run(done chan error) { } if err != nil { - once.Do(func() { done <- err }) + if reportError(&once, done, err) { + return + } t.log.ERROR.Println(err) continue diff --git a/tariff/helper.go b/tariff/helper.go index 7ab33a459..f798868f3 100644 --- a/tariff/helper.go +++ b/tariff/helper.go @@ -3,6 +3,7 @@ package tariff import ( "errors" "strings" + "sync" "time" "github.com/cenkalti/backoff/v4" @@ -90,3 +91,19 @@ func runOrError[T any, I runnable[T]](t I) (*T, error) { return t, nil } + +// reportError reports the first error to done via once and returns true when +// this is the startup failure - i.e. run's first update failed and runOrError +// is about to discard the tariff. The caller must then return so the goroutine +// exits instead of polling the API forever in the background. +// +// Once a tariff has started successfully, once has already fired (with close), +// so reportError is a no-op returning false and the caller keeps retrying +// transient errors as before. +func reportError(once *sync.Once, done chan<- error, err error) (startupFailed bool) { + once.Do(func() { + startupFailed = true + done <- err + }) + return startupFailed +} diff --git a/tariff/helper_test.go b/tariff/helper_test.go index 1a0ed921b..93e7ddee3 100644 --- a/tariff/helper_test.go +++ b/tariff/helper_test.go @@ -2,6 +2,7 @@ package tariff import ( "errors" + "sync" "testing" "time" @@ -85,3 +86,42 @@ func TestRunOrQError(t *testing.T) { require.Nil(t, res) } } + +// leakRunner mirrors the real tariff run loop: its first update always fails +// (e.g. HTTP 429 at startup) and it reports the error via reportError. With the +// fix in place reportError returns true and run returns; without it the loop +// would block on the hourly tick and keep polling forever. +type leakRunner struct { + exited chan struct{} +} + +func (r *leakRunner) run(done chan error) { + defer close(r.exited) + + var once sync.Once + for tick := time.Tick(time.Hour); ; <-tick { + if reportError(&once, done, errors.New("initial failure (e.g. HTTP 429)")) { + return + } + // without the fix the goroutine would block on <-tick here and keep + // hitting the API on every tick - exactly the leak we guard against + } +} + +// TestRunOrErrorStopsGoroutineOnStartupFailure asserts that when the first +// update fails, runOrError both propagates the error and stops the background +// goroutine instead of leaving it to poll the API indefinitely. +func TestRunOrErrorStopsGoroutineOnStartupFailure(t *testing.T) { + r := &leakRunner{exited: make(chan struct{})} + + res, err := runOrError(r) + require.Error(t, err, "startup failure must be propagated") + require.Nil(t, res, "tariff must not be returned when startup fails") + + select { + case <-r.exited: + // goroutine returned - no leak + case <-time.After(time.Second): + t.Fatal("run goroutine still alive after startup failure (goroutine leak)") + } +} diff --git a/tariff/ngeso.go b/tariff/ngeso.go index 1c61df43f..ea014e159 100644 --- a/tariff/ngeso.go +++ b/tariff/ngeso.go @@ -75,7 +75,9 @@ func (t *Ngeso) run(done chan error) { return res, backoffPermanentError(err) }, bo()) if err != nil { - once.Do(func() { done <- err }) + if reportError(&once, done, err) { + return + } t.log.ERROR.Println(err) continue diff --git a/tariff/octopus.go b/tariff/octopus.go index 46cf4d5fc..d81e5b61a 100644 --- a/tariff/octopus.go +++ b/tariff/octopus.go @@ -156,7 +156,9 @@ func (t *Octopus) run(done chan error) { if err := backoff.Retry(func() error { return backoffPermanentError(client.GetJSON(restQueryUri, &res)) }, bo()); err != nil { - once.Do(func() { done <- err }) + if reportError(&once, done, err) { + return + } t.log.ERROR.Println(err) continue diff --git a/tariff/octopusde.go b/tariff/octopusde.go index ba0313e7b..57a98b15a 100644 --- a/tariff/octopusde.go +++ b/tariff/octopusde.go @@ -104,7 +104,9 @@ func (t *OctopusDe) run(done chan error) { rates, err = ratesForAgreement(agr, time.Now()) return backoffPermanentError(err) }, bo()); err != nil { - once.Do(func() { done <- err }) + if reportError(&once, done, err) { + return + } t.log.ERROR.Printf("failed to fetch unit rate forecast: %v", err) continue diff --git a/tariff/ostrom.go b/tariff/ostrom.go index f71b69376..22d339243 100644 --- a/tariff/ostrom.go +++ b/tariff/ostrom.go @@ -207,7 +207,9 @@ func (t *Ostrom) runStatic(done chan error) { for tick := time.Tick(time.Hour); ; <-tick { price, err := t.getFixedPrice() if err != nil { - once.Do(func() { done <- err }) + if reportError(&once, done, err) { + return + } t.log.ERROR.Println(err) continue } @@ -249,7 +251,9 @@ func (t *Ostrom) run(done chan error) { if err := backoff.Retry(func() error { return backoffPermanentError(t.GetJSON(uri, &res)) }, bo()); err != nil { - once.Do(func() { done <- err }) + if reportError(&once, done, err) { + return + } t.log.ERROR.Println(err) continue } diff --git a/tariff/pun.go b/tariff/pun.go index 85c9baaae..dfa16bc20 100644 --- a/tariff/pun.go +++ b/tariff/pun.go @@ -86,7 +86,9 @@ func (t *Pun) run(done chan error) { return res, backoffPermanentError(err) }, bo()) if err != nil { - once.Do(func() { done <- err }) + if reportError(&once, done, err) { + return + } t.log.ERROR.Println(err) continue } @@ -100,7 +102,9 @@ func (t *Pun) run(done chan error) { return res, backoffPermanentError(err) }, bo()) if err != nil && !errors.Is(err, ErrPunDataNotAvailable) { - once.Do(func() { done <- err }) + if reportError(&once, done, err) { + return + } t.log.ERROR.Println(err) continue } diff --git a/tariff/smartenergy.go b/tariff/smartenergy.go index fe3d29084..895629db7 100644 --- a/tariff/smartenergy.go +++ b/tariff/smartenergy.go @@ -56,7 +56,9 @@ func (t *SmartEnergy) run(done chan error) { if err := backoff.Retry(func() error { return backoffPermanentError(client.GetJSON(smartenergy.URI, &res)) }, bo()); err != nil { - once.Do(func() { done <- err }) + if reportError(&once, done, err) { + return + } t.log.ERROR.Println(err) continue diff --git a/tariff/solcast.go b/tariff/solcast.go index eb6bb3ef4..7619764dd 100644 --- a/tariff/solcast.go +++ b/tariff/solcast.go @@ -93,7 +93,9 @@ func (t *Solcast) run(interval time.Duration, done chan error) { uri := fmt.Sprintf("https://api.solcast.com.au/rooftop_sites/%s/forecasts?period=PT30M&format=json", t.site) return backoffPermanentError(t.GetJSON(uri, &res)) }, bo()); err != nil { - once.Do(func() { done <- err }) + if reportError(&once, done, err) { + return + } t.log.ERROR.Println(err) continue } diff --git a/tariff/stekker.go b/tariff/stekker.go index b803ff935..a6eb7ce37 100644 --- a/tariff/stekker.go +++ b/tariff/stekker.go @@ -90,13 +90,17 @@ func (t *Stekker) run(done chan error) { url := fmt.Sprintf("%s?advanced_view=®ion=%s&unit=MWh", stekkerURI, t.region) resp, err := client.Get(url) if err != nil { - once.Do(func() { done <- err }) + if reportError(&once, done, err) { + return + } t.log.ERROR.Println("http error:", err) continue } if resp.StatusCode != http.StatusOK { - once.Do(func() { done <- fmt.Errorf("http status %d", resp.StatusCode) }) + if reportError(&once, done, fmt.Errorf("http status %d", resp.StatusCode)) { + return + } t.log.ERROR.Printf("http status %d", resp.StatusCode) resp.Body.Close() continue @@ -105,7 +109,9 @@ func (t *Stekker) run(done chan error) { doc, err := goquery.NewDocumentFromReader(resp.Body) if err != nil { resp.Body.Close() - once.Do(func() { done <- err }) + if reportError(&once, done, err) { + return + } t.log.ERROR.Println("parse error:", err) continue } @@ -113,7 +119,9 @@ func (t *Stekker) run(done chan error) { val, ok := doc.Find("[data-epex-forecast-graph-data-value]").Attr("data-epex-forecast-graph-data-value") if !ok { - once.Do(func() { done <- fmt.Errorf("no forecast attribute found") }) + if reportError(&once, done, fmt.Errorf("no forecast attribute found")) { + return + } t.log.ERROR.Println("no forecast attribute found") continue } @@ -122,7 +130,9 @@ func (t *Stekker) run(done chan error) { var data []map[string]any if err := json.Unmarshal([]byte(raw), &data); err != nil { - once.Do(func() { done <- err }) + if reportError(&once, done, err) { + return + } t.log.ERROR.Println("unmarshal error:", err) continue } diff --git a/tariff/tariff.go b/tariff/tariff.go index 0add9df1a..6f3069dd3 100644 --- a/tariff/tariff.go +++ b/tariff/tariff.go @@ -108,7 +108,9 @@ func (t *Tariff) run(forecastG func() (string, error), done chan error, interval } return nil }, bo()); err != nil { - once.Do(func() { done <- err }) + if reportError(&once, done, err) { + return + } t.log.ERROR.Println(err) continue diff --git a/tariff/tibber.go b/tariff/tibber.go index 9429a74af..630c1923b 100644 --- a/tariff/tibber.go +++ b/tariff/tibber.go @@ -92,7 +92,9 @@ func (t *Tibber) run(done chan error) { defer cancel() return t.client.Query(ctx, &res, v) }, bo()); err != nil { - once.Do(func() { done <- err }) + if reportError(&once, done, err) { + return + } t.log.ERROR.Println(err) continue