Tariff: stop background routine when startup update fails (#30594)

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

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

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

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

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

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
 		}

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

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
+}

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)")
+	}
+}

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

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

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

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
 		}

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
 		}

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

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
 		}

tariff/stekker.go

@@ -90,13 +90,17 @@ func (t *Stekker) run(done chan error) {
 		url := fmt.Sprintf("%s?advanced_view=&region=%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
 		}

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

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