fix(weather): throttle failed forecast fetches, not only successes

.claude/skills/beacon-collection/SKILL.md

@@ -153,6 +153,19 @@ separate from the current-conditions path.
   least 24 h since the last capture" drifts an hour later every day and eventually lands
   after the subscriber's notify hour, so the digest would read a forecast a day older than
   it needed to be. A calendar day pins the fetch to the first tick after midnight UTC.
+- **The gate throttles failures too, or it throttles nothing.** A capture is only written on
+  success, so on its own the calendar-day gate leaves a location that cannot be fetched due
+  for the rest of the day and retried on every tick — each retry being another
+  `openMeteoMaxAttempts` requests. A per-location marker under
+  `repository.ServiceMetaKeyForecastAttemptPrefix` in `service_meta` counts the day's failed
+  attempts and doubles the wait after each, so tries land at roughly 0, 1, 3, 7 and 15 hours
+  and then stop: the spacing matters because the outages measured against this provider run
+  about three hours per location, and a budget burned in the first few ticks would miss the
+  recovery. A failing location is logged as `deferred=`, never as `skipped=` — the two are
+  opposite states, and conflating them is what made the unthrottled version look healthy.
+  Every uncertainty about the marker resolves to "fetch": unreadable, unparseable, or from
+  another day all mean a fresh budget, because a wrong "yes" costs one request and a wrong
+  "no" is a location that silently stops updating.
 - **Retention keeps one day of slack.** `RemoveForecastDaysBefore` runs on every tick
   whatever the fetches did, with a cutoff of yesterday UTC: offsets run from −12 to +14, so
   one extra day is what makes "past" unambiguous for every subscriber.

cmd/collector/main.go

@@ -116,7 +116,7 @@ func main() {
 
 	runners, err := buildRunners(
 		sourceRepo, historyRepo, rateValueRepo,
-		weatherCityRepo, weatherObsRepo, weatherForecastRepo,
+		weatherCityRepo, weatherObsRepo, weatherForecastRepo, metaRepo,
 		l.WriterAs(internal.LogLevelWarning),
 	)
 	if err != nil {
@@ -267,6 +267,7 @@ func buildRunners(
 	weatherCity *repository.WeatherUserCityRepository,
 	weatherObs *repository.WeatherObservationRepository,
 	weatherForecast *repository.WeatherForecastDayRepository,
+	meta *repository.ServiceMetaRepository,
 	logger io.Writer,
 ) ([]runner, error) {
 	// Passing the proxy URL does not route anything through it: the extractor builds a
@@ -285,7 +286,7 @@ func buildRunners(
 		return nil, errors.Join(err, loginjector.NewTraceError())
 	}
 
-	weatherAgents, err := wireWeather(weatherCity, weatherObs, weatherForecast, logger)
+	weatherAgents, err := wireWeather(weatherCity, weatherObs, weatherForecast, meta, logger)
 	if err != nil {
 		return nil, errors.Join(err, loginjector.NewTraceError())
 	}
@@ -309,6 +310,7 @@ func wireWeather(
 	weatherCity *repository.WeatherUserCityRepository,
 	weatherObs *repository.WeatherObservationRepository,
 	weatherForecast *repository.WeatherForecastDayRepository,
+	meta *repository.ServiceMetaRepository,
 	logger io.Writer,
 ) ([]runner, error) {
 	openMeteoProvider, err := weatherinfra.NewOpenMeteo("", logger)
@@ -326,7 +328,7 @@ func wireWeather(
 	}
 
 	forecastAgent, err := collection.NewWeatherForecastAgent(
-		openMeteoProvider, weatherCity, weatherForecast,
+		openMeteoProvider, weatherCity, weatherForecast, meta,
 		logger,
 	)
 	if err != nil {

cmd/collector/main_test.go

@@ -33,8 +33,10 @@ func TestWireWeather(t *testing.T) {
 		require.NoError(t, err)
 		forecastRepo, err := repository.NewWeatherForecastDayRepository(nil)
 		require.NoError(t, err)
+		metaRepo, err := repository.NewServiceMetaRepository(nil)
+		require.NoError(t, err)
 
-		agents, err := wireWeather(cityRepo, obsRepo, forecastRepo, nil)
+		agents, err := wireWeather(cityRepo, obsRepo, forecastRepo, metaRepo, nil)
 		require.NoError(t, err)
 		require.Len(t, agents, 2, "current conditions and the long-range forecast are separate runners")
 		for i, agent := range agents {
@@ -54,8 +56,10 @@ func TestWireWeather(t *testing.T) {
 		require.NoError(t, err)
 		forecastRepo, err := repository.NewWeatherForecastDayRepository(nil)
 		require.NoError(t, err)
+		metaRepo, err := repository.NewServiceMetaRepository(nil)
+		require.NoError(t, err)
 
-		agents, err := wireWeather(cityRepo, obsRepo, forecastRepo, nil)
+		agents, err := wireWeather(cityRepo, obsRepo, forecastRepo, metaRepo, nil)
 		require.NoError(t, err)
 		assert.NotEmpty(t, agents)
 	})
@@ -79,8 +83,10 @@ func TestWeatherIgnoresProxyEnv(t *testing.T) {
 	require.NoError(t, err)
 	forecastRepo, err := repository.NewWeatherForecastDayRepository(nil)
 	require.NoError(t, err)
+	metaRepo, err := repository.NewServiceMetaRepository(nil)
+	require.NoError(t, err)
 
-	agents, err := wireWeather(cityRepo, obsRepo, forecastRepo, nil)
+	agents, err := wireWeather(cityRepo, obsRepo, forecastRepo, metaRepo, nil)
 	require.NoError(t, err, "a proxy setting in the environment must be inert for weather")
 	assert.NotEmpty(t, agents)
 }

internal/application/collection/weatherforecastagent.go

@@ -5,10 +5,13 @@ import (
 	"errors"
 	"fmt"
 	"io"
+	"strconv"
+	"strings"
 	"time"
 
 	"github.com/prorochestvo/loginjector"
 	"github.com/seilbekskindirov/beacon/internal/domain"
+	"github.com/seilbekskindirov/beacon/internal/repository"
 )
 
 // WeatherForecastAgent collects the multi-week daily forecast from Open-Meteo for every
@@ -25,19 +28,21 @@ type WeatherForecastAgent struct {
 	provider weatherRangeProvider
 	cityRepo weatherCollectionCityRepo
 	dayRepo  weatherForecastDayRepo
+	meta     metaRepository
 	logger   io.Writer
 }
 
-// NewWeatherForecastAgent constructs a WeatherForecastAgent. provider, cityRepo and dayRepo
-// are all required; a nil logger discards output.
+// NewWeatherForecastAgent constructs a WeatherForecastAgent. provider, cityRepo, dayRepo
+// and meta are all required; a nil logger discards output.
 func NewWeatherForecastAgent(
 	provider weatherRangeProvider,
 	cityRepo weatherCollectionCityRepo,
 	dayRepo weatherForecastDayRepo,
+	meta metaRepository,
 	logger io.Writer,
 ) (*WeatherForecastAgent, error) {
-	if provider == nil || cityRepo == nil || dayRepo == nil {
-		return nil, errors.New("weather forecast agent: provider, cityRepo, and dayRepo are all required")
+	if provider == nil || cityRepo == nil || dayRepo == nil || meta == nil {
+		return nil, errors.New("weather forecast agent: provider, cityRepo, dayRepo, and meta are all required")
 	}
 	if logger == nil {
 		logger = io.Discard
@@ -46,6 +51,7 @@ func NewWeatherForecastAgent(
 		provider: provider,
 		cityRepo: cityRepo,
 		dayRepo:  dayRepo,
+		meta:     meta,
 		logger:   logger,
 	}, nil
 }
@@ -62,18 +68,27 @@ func (a *WeatherForecastAgent) Run(ctx context.Context) error {
 
 	now := time.Now().UTC()
 	var errs []error
-	var fetched, skipped, failed int
+	var fetched, skipped, deferred, failed int
 	total := len(locations)
 
 	for _, loc := range locations {
-		if !a.isDue(ctx, loc.LocationID, now) {
+		switch a.dueness(ctx, loc.LocationID, now) {
+		case forecastFetched:
 			skipped++
 			continue
+		case forecastBackingOff:
+			// Counted apart from skipped on purpose. "Already have today's" and "failing and
+			// waiting out its backoff" are opposite states, and a log that spells both
+			// skipped is the log that made this bug invisible.
+			deferred++
+			continue
+		case forecastDue:
 		}
 
 		days, fetchErr := a.provider.ForecastRange(ctx, loc.Latitude, loc.Longitude)
 		if fetchErr != nil {
 			failed++
+			a.recordFailedAttempt(ctx, loc.LocationID, now)
 			fmt.Fprintf(a.logger, "weather forecast: location %s: fetch error: %v\n", loc.LocationID, fetchErr)
 			errs = append(errs, fmt.Errorf("location %s: forecast range: %w", loc.LocationID, fetchErr))
 			continue
@@ -87,14 +102,16 @@ func (a *WeatherForecastAgent) Run(ctx context.Context) error {
 		//nolint:contextcheck // the detached context is the point; see the comment above
 		if retainErr := a.dayRepo.RetainWeatherForecastDays(context.Background(), days); retainErr != nil {
 			failed++
+			a.recordFailedAttempt(ctx, loc.LocationID, now)
 			fmt.Fprintf(a.logger, "weather forecast: location %s: retain error: %v\n", loc.LocationID, retainErr)
 			errs = append(errs, fmt.Errorf("location %s: retain forecast: %w", loc.LocationID, retainErr))
 			continue
 		}
 		fetched++
 	}
 
-	fmt.Fprintf(a.logger, "weather forecast: fetched=%d skipped=%d failed=%d total=%d\n", fetched, skipped, failed, total)
+	fmt.Fprintf(a.logger, "weather forecast: fetched=%d skipped=%d deferred=%d failed=%d total=%d\n",
+		fetched, skipped, deferred, failed, total)
 
 	// A day is kept until it is behind every subscriber, not merely behind UTC. Offsets run
 	// from -12 to +14, so one whole day of slack is what makes "past" unambiguous; the cost
@@ -108,22 +125,143 @@ func (a *WeatherForecastAgent) Run(ctx context.Context) error {
 	return errors.Join(errs...)
 }
 
-// isDue reports whether the location still needs its fetch for the current UTC day.
+// forecastDueness is what one location owes the current tick.
+type forecastDueness uint8
+
+const (
+	// forecastDue means the location should be fetched now.
+	forecastDue forecastDueness = iota
+	// forecastFetched means today's forecast is already stored.
+	forecastFetched
+	// forecastBackingOff means today's fetch has failed and the location is waiting out its
+	// backoff, or has spent the day's budget of attempts.
+	forecastBackingOff
+)
+
+// weatherForecastMaxDailyAttempts caps how many times one location is fetched in a UTC day
+// while it keeps failing. Each attempt is itself up to openMeteoMaxAttempts HTTP requests, so
+// five here is at most twenty-five requests a day for a location that never answers, against
+// the twenty-four ticks times five that an ungated retry costs.
+const weatherForecastMaxDailyAttempts = 5
+
+// weatherForecastRetryBase is the wait after the first failed fetch of the day; each further
+// failure doubles it. With the cap above, attempts land at roughly 0, 1, 3, 7 and 15 hours
+// in — spread across the day rather than burned in the first five ticks, which matters
+// because the outages measured against this provider run about three hours per location.
 //
-// The gate is a calendar-day comparison rather than "at least 24 h since the last capture".
-// Against an hourly cron the elapsed-time form drifts an hour later every day and eventually
-// lands after the subscriber's notify hour, so the digest would read a forecast a day older
-// than it needed to be; a calendar day pins the fetch to the first tick after midnight UTC
-// and stays there.
+// To make retries denser or sparser, move this constant; to change how many there are, move
+// the cap. Nothing else reads either.
+const weatherForecastRetryBase = time.Hour
+
+// dueness reports what the location owes this tick.
 //
-// A read failure counts as due. ErrNotFound means the location has never been fetched, and
-// anything else must not be allowed to skip a location permanently.
-func (a *WeatherForecastAgent) isDue(ctx context.Context, locationID string, now time.Time) bool {
+// The primary gate is a calendar-day comparison rather than "at least 24 h since the last
+// capture". Against an hourly cron the elapsed-time form drifts an hour later every day and
+// eventually lands after the subscriber's notify hour, so the digest would read a forecast a
+// day older than it needed to be; a calendar day pins the fetch to the first tick after
+// midnight UTC and stays there.
+//
+// A capture read that fails is not treated as "already fetched" — ErrNotFound means the
+// location has never been fetched, and a storage fault must not skip a location permanently.
+// It does still fall through to the attempt budget, because a persistent read fault would
+// otherwise convert a storage problem into upstream traffic.
+func (a *WeatherForecastAgent) dueness(ctx context.Context, locationID string, now time.Time) forecastDueness {
 	last, err := a.dayRepo.ObtainLatestForecastCapture(ctx, locationID, domain.ProviderOpenMeteo)
-	if err != nil {
+	if err == nil && last.UTC().Format(time.DateOnly) == now.Format(time.DateOnly) {
+		return forecastFetched
+	}
+	if !a.retryWindowOpen(ctx, locationID, now) {
+		return forecastBackingOff
+	}
+	return forecastDue
+}
+
+// retryWindowOpen reports whether the location's attempt budget and backoff allow a fetch now.
+//
+// Every uncertainty resolves to true. A marker that cannot be read, cannot be parsed, or
+// belongs to another day must not be what stops collection: the worst case of a wrong "true"
+// is one extra request, and the worst case of a wrong "false" is a location that silently
+// never updates.
+func (a *WeatherForecastAgent) retryWindowOpen(ctx context.Context, locationID string, now time.Time) bool {
+	raw, ok, err := a.meta.ObtainServiceMeta(ctx, forecastAttemptKey(locationID))
+	if err != nil || !ok {
+		return true
+	}
+
+	marker, parseErr := parseForecastAttempt(raw)
+	if parseErr != nil || marker.day != now.Format(time.DateOnly) {
 		return true
 	}
-	return last.UTC().Format(time.DateOnly) != now.Format(time.DateOnly)
+	if marker.count >= weatherForecastMaxDailyAttempts {
+		return false
+	}
+	return !now.Before(marker.lastAt.Add(forecastRetryWait(marker.count)))
+}
+
+// recordFailedAttempt increments the location's attempt count for today. A failure to write
+// the marker is logged and swallowed: the cost is that this location keeps the old
+// every-tick behaviour until the write succeeds, which is strictly better than losing the
+// fetch itself to a bookkeeping error.
+func (a *WeatherForecastAgent) recordFailedAttempt(ctx context.Context, locationID string, now time.Time) {
+	today := now.Format(time.DateOnly)
+
+	count := 0
+	if raw, ok, err := a.meta.ObtainServiceMeta(ctx, forecastAttemptKey(locationID)); err == nil && ok {
+		if marker, parseErr := parseForecastAttempt(raw); parseErr == nil && marker.day == today {
+			count = marker.count
+		}
+	}
+
+	value := formatForecastAttempt(forecastAttemptMarker{day: today, count: count + 1, lastAt: now})
+	if err := a.meta.RetainServiceMeta(ctx, forecastAttemptKey(locationID), value); err != nil {
+		fmt.Fprintf(a.logger, "weather forecast: location %s: record attempt: %v\n", locationID, err)
+	}
+}
+
+// forecastRetryWait returns how long to wait after the given number of failures today.
+func forecastRetryWait(failures int) time.Duration {
+	if failures < 1 {
+		return 0
+	}
+	return weatherForecastRetryBase << (failures - 1)
+}
+
+// forecastAttemptKey is the service_meta key holding one location's attempts for today. The
+// row outlives a location that stops being subscribed; at a few dozen bytes each that is
+// cheaper to leave than to reconcile.
+func forecastAttemptKey(locationID string) string {
+	return repository.ServiceMetaKeyForecastAttemptPrefix + locationID
+}
+
+// forecastAttemptMarker is one location's failed-fetch state for a single UTC day.
+type forecastAttemptMarker struct {
+	day    string // YYYY-MM-DD, UTC — a marker from another day is a fresh budget
+	count  int
+	lastAt time.Time
+}
+
+// formatForecastAttempt encodes a marker as "<YYYY-MM-DD>|<count>|<RFC3339>".
+func formatForecastAttempt(m forecastAttemptMarker) string {
+	return fmt.Sprintf("%s|%d|%s", m.day, m.count, m.lastAt.UTC().Format(time.RFC3339))
+}
+
+// parseForecastAttempt decodes what formatForecastAttempt wrote. Every caller treats an
+// error as "no usable marker" rather than as a failure.
+func parseForecastAttempt(raw string) (forecastAttemptMarker, error) {
+	parts := strings.Split(raw, "|")
+	if len(parts) != 3 {
+		return forecastAttemptMarker{}, fmt.Errorf("forecast attempt marker: want 3 fields, got %d", len(parts))
+	}
+
+	count, err := strconv.Atoi(parts[1])
+	if err != nil {
+		return forecastAttemptMarker{}, fmt.Errorf("forecast attempt marker: count: %w", err)
+	}
+	lastAt, err := time.Parse(time.RFC3339, parts[2])
+	if err != nil {
+		return forecastAttemptMarker{}, fmt.Errorf("forecast attempt marker: last attempt: %w", err)
+	}
+	return forecastAttemptMarker{day: parts[0], count: count, lastAt: lastAt}, nil
 }
 
 // weatherRangeProvider fetches a multi-week daily forecast for the given coordinates.