From 7516ec32561a6d84f15285d1553536990dda981e Mon Sep 17 00:00:00 2001 From: prorochestvo Date: Sun, 23 Aug 2026 10:26:47 +0500 Subject: [PATCH] fix(weather): throttle failed forecast fetches, not only successes MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit A capture is written only on success, so the calendar-day gate never closed for a location that could not be fetched: it stayed due for the rest of the UTC day and was retried on every tick, each retry being another openMeteoMaxAttempts requests. Against the outage pattern already measured for this provider — 105 of 177 fetches meeting a 503 over five days, episodes about three hours per location — the planned one weighted call per location per day became up to a hundred and twenty on an hourly cron. A persistent read fault on ObtainLatestForecastCapture did the same, converting a storage problem into upstream traffic. Count the day's failed attempts per location in service_meta and double the wait after each, so tries land at roughly 0, 1, 3, 7 and 15 hours and then stop. The spacing is the point: a flat budget burned in the first few ticks would miss a three-hour outage recovering. Move weatherForecastRetryBase to change the density and weatherForecastMaxDailyAttempts to change the count; nothing else reads either. Every uncertainty about the marker resolves to "fetch" — unreadable, unparseable, or from another day all mean a fresh budget, since a wrong yes costs one request while a wrong no is a location that silently stops updating. A deferred location is logged as deferred=, never skipped=. Conflating "already have today's" with "failing and waiting" is what let the unthrottled version read as healthy. Refs: #132 Co-Authored-By: Claude Opus 5 Claude-Session: https://claude.ai/code/session_01U6jNuzK5PZrhDWBxB3t2gY --- .claude/skills/beacon-collection/SKILL.md | 13 ++ cmd/collector/main.go | 8 +- cmd/collector/main_test.go | 12 +- .../collection/weatherforecastagent.go | 174 ++++++++++++-- .../collection/weatherforecastagent_test.go | 216 +++++++++++++++++- internal/repository/servicemeta.go | 6 + 6 files changed, 399 insertions(+), 30 deletions(-) diff --git a/.claude/skills/beacon-collection/SKILL.md b/.claude/skills/beacon-collection/SKILL.md index 0f0a5c8..cec6a34 100644 --- a/.claude/skills/beacon-collection/SKILL.md +++ b/.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. diff --git a/cmd/collector/main.go b/cmd/collector/main.go index 0978f19..ca813e4 100644 --- a/cmd/collector/main.go +++ b/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 { diff --git a/cmd/collector/main_test.go b/cmd/collector/main_test.go index bd6a57b..88de610 100644 --- a/cmd/collector/main_test.go +++ b/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) } diff --git a/internal/application/collection/weatherforecastagent.go b/internal/application/collection/weatherforecastagent.go index 43377c6..277e336 100644 --- a/internal/application/collection/weatherforecastagent.go +++ b/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,6 +102,7 @@ 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 @@ -94,7 +110,8 @@ func (a *WeatherForecastAgent) Run(ctx context.Context) error { 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 "||". +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. diff --git a/internal/application/collection/weatherforecastagent_test.go b/internal/application/collection/weatherforecastagent_test.go index fdc6626..42870e8 100644 --- a/internal/application/collection/weatherforecastagent_test.go +++ b/internal/application/collection/weatherforecastagent_test.go @@ -26,26 +26,32 @@ func TestNewWeatherForecastAgent(t *testing.T) { t.Run("valid construction", func(t *testing.T) { t.Parallel() - a, err := NewWeatherForecastAgent(&mockWeatherRangeProvider{}, &mockWeatherCityRepo{}, &mockWeatherForecastDayRepo{}, io.Discard) + a, err := NewWeatherForecastAgent(&mockWeatherRangeProvider{}, &mockWeatherCityRepo{}, &mockWeatherForecastDayRepo{}, newFakeMetaRepo(), io.Discard) require.NoError(t, err) require.NotNil(t, a) }) t.Run("nil provider returns error", func(t *testing.T) { t.Parallel() - _, err := NewWeatherForecastAgent(nil, &mockWeatherCityRepo{}, &mockWeatherForecastDayRepo{}, io.Discard) + _, err := NewWeatherForecastAgent(nil, &mockWeatherCityRepo{}, &mockWeatherForecastDayRepo{}, newFakeMetaRepo(), io.Discard) require.Error(t, err) }) t.Run("nil cityRepo returns error", func(t *testing.T) { t.Parallel() - _, err := NewWeatherForecastAgent(&mockWeatherRangeProvider{}, nil, &mockWeatherForecastDayRepo{}, io.Discard) + _, err := NewWeatherForecastAgent(&mockWeatherRangeProvider{}, nil, &mockWeatherForecastDayRepo{}, newFakeMetaRepo(), io.Discard) require.Error(t, err) }) t.Run("nil dayRepo returns error", func(t *testing.T) { t.Parallel() - _, err := NewWeatherForecastAgent(&mockWeatherRangeProvider{}, &mockWeatherCityRepo{}, nil, io.Discard) + _, err := NewWeatherForecastAgent(&mockWeatherRangeProvider{}, &mockWeatherCityRepo{}, nil, newFakeMetaRepo(), io.Discard) + require.Error(t, err) + }) + + t.Run("nil meta returns error", func(t *testing.T) { + t.Parallel() + _, err := NewWeatherForecastAgent(&mockWeatherRangeProvider{}, &mockWeatherCityRepo{}, &mockWeatherForecastDayRepo{}, nil, io.Discard) require.Error(t, err) }) } @@ -171,12 +177,210 @@ func TestWeatherForecastAgent_Run(t *testing.T) { &mockWeatherRangeProvider{days: []domain.WeatherForecastDay{{ForecastDate: "2026-08-21"}}}, &mockWeatherCityRepo{locations: locations("loc1")}, dayRepo, + newFakeMetaRepo(), &log, ) require.NoError(t, err) require.NoError(t, a.Run(t.Context())) - assert.Contains(t, log.String(), "weather forecast: fetched=1 skipped=0 failed=0 total=1") + assert.Contains(t, log.String(), "weather forecast: fetched=1 skipped=0 deferred=0 failed=0 total=1") + }) +} + +func TestWeatherForecastAgentAttemptBudget(t *testing.T) { + t.Parallel() + + key := forecastAttemptKey("loc1") + + // agentWith builds an agent over one always-failing location and the given marker state. + agentWith := func(t *testing.T, meta *fakeMetaRepo, log io.Writer) (*WeatherForecastAgent, *mockWeatherRangeProvider) { + t.Helper() + provider := &mockWeatherRangeProvider{failOnLat: 1} + a, err := NewWeatherForecastAgent( + provider, + &mockWeatherCityRepo{locations: locations("loc1")}, + &mockWeatherForecastDayRepo{captureErr: internal.ErrNotFound}, + meta, + log, + ) + require.NoError(t, err) + return a, provider + } + + t.Run("a failed fetch records the attempt", func(t *testing.T) { + t.Parallel() + meta := newFakeMetaRepo() + a, provider := agentWith(t, meta, io.Discard) + + require.Error(t, a.Run(t.Context())) + assert.Equal(t, 1, provider.calls) + + marker, err := parseForecastAttempt(meta.values[key]) + require.NoError(t, err, "the failure must leave a parseable marker") + assert.Equal(t, time.Now().UTC().Format(time.DateOnly), marker.day) + assert.Equal(t, 1, marker.count) + }) + + t.Run("a retain failure records the attempt too", func(t *testing.T) { + t.Parallel() + // The fetch was paid for either way; only the store failed. + meta := newFakeMetaRepo() + a, err := NewWeatherForecastAgent( + &mockWeatherRangeProvider{days: []domain.WeatherForecastDay{{ForecastDate: "2026-08-21"}}}, + &mockWeatherCityRepo{locations: locations("loc1")}, + &mockWeatherForecastDayRepo{captureErr: internal.ErrNotFound, retainErr: errors.New("disk on fire")}, + meta, + io.Discard, + ) + require.NoError(t, err) + + require.Error(t, a.Run(t.Context())) + assert.NotEmpty(t, meta.values[key]) + }) + + t.Run("a spent budget defers the location instead of refetching", func(t *testing.T) { + t.Parallel() + // The whole point: before this, a location that could not be fetched stayed due and + // was retried on every tick for the rest of the day. + var log strings.Builder + meta := newFakeMetaRepo() + meta.values[key] = formatForecastAttempt(forecastAttemptMarker{ + day: time.Now().UTC().Format(time.DateOnly), + count: weatherForecastMaxDailyAttempts, + lastAt: time.Now().UTC().Add(-24 * time.Hour), + }) + a, provider := agentWith(t, meta, &log) + + require.NoError(t, a.Run(t.Context())) + assert.Zero(t, provider.calls, "the budget is spent; nothing may reach the provider") + assert.Contains(t, log.String(), "deferred=1") + assert.Contains(t, log.String(), "skipped=0", "backing off is not the same state as already fetched") }) + + t.Run("a marker from another day is a fresh budget", func(t *testing.T) { + t.Parallel() + meta := newFakeMetaRepo() + meta.values[key] = formatForecastAttempt(forecastAttemptMarker{ + day: time.Now().UTC().AddDate(0, 0, -1).Format(time.DateOnly), + count: weatherForecastMaxDailyAttempts, + lastAt: time.Now().UTC().Add(-24 * time.Hour), + }) + a, provider := agentWith(t, meta, io.Discard) + + require.Error(t, a.Run(t.Context())) + assert.Equal(t, 1, provider.calls) + }) + + t.Run("a marker that cannot be read or parsed never blocks collection", func(t *testing.T) { + t.Parallel() + for name, meta := range map[string]*fakeMetaRepo{ + "read fails": {values: map[string]string{}, readErr: errors.New("meta unavailable")}, + "garbage": {values: map[string]string{key: "last tuesday"}}, + } { + t.Run(name, func(t *testing.T) { + t.Parallel() + a, provider := agentWith(t, meta, io.Discard) + require.Error(t, a.Run(t.Context())) + assert.Equal(t, 1, provider.calls, "bookkeeping must not be what stops a fetch") + }) + } + }) + + t.Run("today's stored forecast still wins over any marker", func(t *testing.T) { + t.Parallel() + meta := newFakeMetaRepo() + provider := &mockWeatherRangeProvider{} + a, err := NewWeatherForecastAgent( + provider, + &mockWeatherCityRepo{locations: locations("loc1")}, + &mockWeatherForecastDayRepo{capture: time.Now().UTC()}, + meta, + io.Discard, + ) + require.NoError(t, err) + + require.NoError(t, a.Run(t.Context())) + assert.Zero(t, provider.calls) + assert.Empty(t, meta.values[key], "a location that never failed writes no marker") + }) +} + +func TestWeatherForecastAgentRetryWindow(t *testing.T) { + t.Parallel() + + // retryWindowOpen takes now explicitly, so the timing cases need no clock seam. + now := time.Date(2026, 8, 23, 12, 0, 0, 0, time.UTC) + key := forecastAttemptKey("loc1") + + agentWithMarker := func(t *testing.T, marker forecastAttemptMarker) *WeatherForecastAgent { + t.Helper() + meta := newFakeMetaRepo() + meta.values[key] = formatForecastAttempt(marker) + a, err := NewWeatherForecastAgent( + &mockWeatherRangeProvider{}, &mockWeatherCityRepo{}, &mockWeatherForecastDayRepo{}, meta, io.Discard) + require.NoError(t, err) + return a + } + + cases := []struct { + name string + count int + since time.Duration // how long ago the last attempt was + open bool + }{ + {"one failure, half an hour ago", 1, 30 * time.Minute, false}, + {"one failure, an hour ago", 1, time.Hour, true}, + {"two failures, an hour ago", 2, time.Hour, false}, + {"two failures, two hours ago", 2, 2 * time.Hour, true}, + {"three failures, three hours ago", 3, 3 * time.Hour, false}, + {"three failures, four hours ago", 3, 4 * time.Hour, true}, + {"budget spent, however long ago", weatherForecastMaxDailyAttempts, 24 * time.Hour, false}, + } + for _, c := range cases { + t.Run(c.name, func(t *testing.T) { + t.Parallel() + a := agentWithMarker(t, forecastAttemptMarker{ + day: now.Format(time.DateOnly), + count: c.count, + lastAt: now.Add(-c.since), + }) + assert.Equal(t, c.open, a.retryWindowOpen(t.Context(), "loc1", now)) + }) + } +} + +func TestForecastAttemptMarker(t *testing.T) { + t.Parallel() + + t.Run("a marker round-trips", func(t *testing.T) { + t.Parallel() + want := forecastAttemptMarker{ + day: "2026-08-23", + count: 3, + lastAt: time.Date(2026, 8, 23, 7, 15, 0, 0, time.UTC), + } + got, err := parseForecastAttempt(formatForecastAttempt(want)) + require.NoError(t, err) + assert.Equal(t, want, got) + }) + + for _, raw := range []string{"", "2026-08-23", "2026-08-23|two|2026-08-23T07:15:00Z", "2026-08-23|1|yesterday"} { + t.Run("malformed: "+raw, func(t *testing.T) { + t.Parallel() + _, err := parseForecastAttempt(raw) + require.Error(t, err) + }) + } +} + +func TestForecastRetryWait(t *testing.T) { + t.Parallel() + + // Doubling from the base, so the day's attempts land at roughly 0, 1, 3, 7 and 15 hours. + assert.Zero(t, forecastRetryWait(0)) + assert.Equal(t, weatherForecastRetryBase, forecastRetryWait(1)) + assert.Equal(t, 2*weatherForecastRetryBase, forecastRetryWait(2)) + assert.Equal(t, 4*weatherForecastRetryBase, forecastRetryWait(3)) + assert.Equal(t, 8*weatherForecastRetryBase, forecastRetryWait(4)) } // locations builds the distinct-location rows the collector iterates, one per id, each with @@ -196,7 +400,7 @@ func locations(ids ...string) []domain.WeatherUserCity { // newForecastAgent constructs an agent with a discarding logger. func newForecastAgent(t *testing.T, provider weatherRangeProvider, cityRepo weatherCollectionCityRepo, dayRepo weatherForecastDayRepo) *WeatherForecastAgent { t.Helper() - a, err := NewWeatherForecastAgent(provider, cityRepo, dayRepo, io.Discard) + a, err := NewWeatherForecastAgent(provider, cityRepo, dayRepo, newFakeMetaRepo(), io.Discard) require.NoError(t, err) return a } diff --git a/internal/repository/servicemeta.go b/internal/repository/servicemeta.go index 8dd00c5..97023b3 100644 --- a/internal/repository/servicemeta.go +++ b/internal/repository/servicemeta.go @@ -96,4 +96,10 @@ const ( // ServiceMetaKeyLastVacuum records when VACUUM last completed. It gates the cadence, // so it is written only after a successful run. ServiceMetaKeyLastVacuum = "last_vacuum_at" + + // ServiceMetaKeyForecastAttemptPrefix prefixes one key per location, holding that + // location's long-range fetch attempts for the current UTC day. It bounds what a + // provider outage costs: without it a location that cannot be fetched stays due all day + // and is retried on every tick. + ServiceMetaKeyForecastAttemptPrefix = "weather_forecast_attempt:" )