From 27506168f854f5fe6bcc613725b2427be2172c3b Mon Sep 17 00:00:00 2001 From: Eric Rakestraw Date: Thu, 2 Jul 2026 11:33:16 -0500 Subject: [PATCH] Implement warmup and fetch retry in the weatherapi adapter --- docs/integrations/weatherapi.md | 22 ++- docs/troubleshooting.md | 9 +- internal/adapters/weatherapi/client.go | 205 ++++++++++++++++++-- internal/adapters/weatherapi/client_test.go | 203 ++++++++++++++++++- 4 files changed, 409 insertions(+), 30 deletions(-) diff --git a/docs/integrations/weatherapi.md b/docs/integrations/weatherapi.md index 71bda36..0a8d9f7 100644 --- a/docs/integrations/weatherapi.md +++ b/docs/integrations/weatherapi.md @@ -18,6 +18,17 @@ fail before any HTTP request when the base URL is empty or not absolute. The HTTP client uses `weather_api.timeout`. +Before fetching bundle sources, the adapter performs a warmup `GET` to +`/conditions/current` with the same query parameters as the current-conditions +source request. This is a temporary connectivity check for VPN wake-up behavior +until the upstream service provides a dedicated health endpoint. A successful +warmup requires a 2xx response whose body can be read; the adapter does not +decode or validate the response envelope during warmup. + +Warmup attempts, warmup delay, source-fetch retry attempts, and source-fetch +retry delay are internal adapter defaults. They are not configuration-file +fields or CLI flags yet. `weather_api.timeout` applies to each HTTP attempt. + ## Response Envelope Every response used by the adapter must be JSON with a top-level `data` field: @@ -45,6 +56,12 @@ Malformed JSON envelopes, non-2xx statuses, and response read failures include endpoint context in returned errors. Decode errors include source context when they fail the fetch; optional malformed sources follow the missing-source policy. +Source-fetch transport failures and retryable HTTP statuses are retried before +the adapter returns an error. Retryable statuses are `408`, `429`, `500`, +`502`, `503`, and `504`. Non-retryable statuses, malformed JSON envelopes, +missing `data`, `data: null` missing-source outcomes, and source decode errors +are not retried. + ## Query Parameters The adapter sends these query parameters: @@ -116,7 +133,10 @@ bundle/debug artifacts, but prompt-facing SPC module output omits geometry. ## Endpoints Used -The adapter fetches these endpoints once per bundle: +The adapter warms up `/conditions/current` once before bundle fetching begins, +with retries if needed. It then fetches these source endpoints once per bundle, +except when a source request is retried after a transient transport or server +failure: - `/observations` - `/conditions/current` diff --git a/docs/troubleshooting.md b/docs/troubleshooting.md index 73657a4..e9c0aae 100644 --- a/docs/troubleshooting.md +++ b/docs/troubleshooting.md @@ -74,7 +74,7 @@ Relevant docs: [CLI reference](cli.md). Symptom: generation fails with `fetch /...`, an HTTP status, or request context. Likely cause: the configured Weather API endpoint is unreachable, returned a -non-2xx response, or returned an invalid response envelope. +non-2xx response after retries, or returned an invalid response envelope. Diagnostic: @@ -83,8 +83,11 @@ weatherreporter generate daily --config ./config.yml --date 2026-05-29 ``` Safe fix: verify `weather_api.base_url`, network access, and the Weather API -service response. The adapter fetches `/observations`, `/conditions/current`, -`/forecast/hourly`, `/forecast/narrative`, `/alerts/active`, and `/discussion`. +service response. The adapter first warms up `/conditions/current`, then fetches +`/observations`, `/conditions/current`, `/forecast/hourly`, +`/forecast/narrative`, `/alerts/active`, `/discussion`, +`/weatherstories/latest`, and `/outlooks/convective`. Transient VPN wake-up +failures and retryable upstream statuses are retried automatically. Relevant docs: [Configuration reference](config.md). diff --git a/internal/adapters/weatherapi/client.go b/internal/adapters/weatherapi/client.go index 70261f1..d3f7edb 100644 --- a/internal/adapters/weatherapi/client.go +++ b/internal/adapters/weatherapi/client.go @@ -24,6 +24,12 @@ import ( const ( convectiveOutlooksEndpoint = "/outlooks/convective" sourceSPCConvectiveOutlooks = "spc_convective_outlooks" + + defaultWarmupEndpoint = "/conditions/current" + defaultWarmupAttempts = 3 + defaultWarmupDelay = time.Second + defaultFetchAttempts = 2 + defaultFetchRetryDelay = time.Second ) type Client struct { @@ -35,6 +41,12 @@ type Client struct { precision int missingSource config.MissingSourceConfig now func() time.Time + + warmupEndpoint string + warmupAttempts int + warmupDelay time.Duration + fetchAttempts int + fetchRetryDelay time.Duration } type Option func(*Client) @@ -80,7 +92,12 @@ func New(cfg config.Config, opts ...Option) (*Client, error) { Default: cfg.MissingSource.Default, Sources: cfg.MissingSource.Sources, }, - now: time.Now, + now: time.Now, + warmupEndpoint: defaultWarmupEndpoint, + warmupAttempts: defaultWarmupAttempts, + warmupDelay: defaultWarmupDelay, + fetchAttempts: defaultFetchAttempts, + fetchRetryDelay: defaultFetchRetryDelay, } for _, opt := range opts { opt(client) @@ -89,6 +106,10 @@ func New(cfg config.Config, opts ...Option) (*Client, error) { } func (c *Client) FetchBundle(ctx context.Context) (*weatherdata.Bundle, error) { + if err := c.warmup(ctx); err != nil { + return nil, err + } + fetchedAt := c.now() builder := bundleBuilder{ client: c, @@ -397,24 +418,9 @@ type envelope struct { } func (c *Client) fetch(ctx context.Context, sourceName string, endpoint string, opts queryOptions) (json.RawMessage, weatherdata.Source, error) { - reqURL := c.endpointURL(endpoint, opts) - req, err := http.NewRequestWithContext(ctx, http.MethodGet, reqURL.String(), nil) + reqURL, body, err := c.fetchHTTP(ctx, endpoint, opts) if err != nil { - return nil, weatherdata.Source{}, fmt.Errorf("create request for %s: %w", endpoint, err) - } - - resp, err := c.httpClient.Do(req) - if err != nil { - return nil, weatherdata.Source{}, fmt.Errorf("fetch %s: %w", endpoint, err) - } - defer resp.Body.Close() - - body, err := io.ReadAll(io.LimitReader(resp.Body, 10<<20)) - if err != nil { - return nil, weatherdata.Source{}, fmt.Errorf("read %s response: %w", endpoint, err) - } - if resp.StatusCode < 200 || resp.StatusCode >= 300 { - return nil, weatherdata.Source{}, fmt.Errorf("fetch %s: unexpected HTTP status %d: %s", endpoint, resp.StatusCode, strings.TrimSpace(string(body))) + return nil, weatherdata.Source{}, err } var env envelope @@ -440,6 +446,169 @@ func (c *Client) fetch(ctx context.Context, sourceName string, endpoint string, return env.Data, source, nil } +func (c *Client) warmup(ctx context.Context) error { + endpoint := c.warmupEndpoint + if strings.TrimSpace(endpoint) == "" { + endpoint = defaultWarmupEndpoint + } + attempts := positiveAttemptCount(c.warmupAttempts) + var lastErr error + for attempt := 1; attempt <= attempts; attempt++ { + if err := ctx.Err(); err != nil { + return fmt.Errorf("warm up weather API via %s: %w", endpoint, err) + } + if err := c.warmupOnce(ctx, endpoint); err != nil { + lastErr = err + } else { + return nil + } + if attempt == attempts { + break + } + if err := waitForRetry(ctx, c.warmupDelay); err != nil { + return fmt.Errorf("warm up weather API via %s after %d attempt(s): %w", endpoint, attempt, err) + } + } + return fmt.Errorf("warm up weather API via %s failed after %d attempts: %w", endpoint, attempts, lastErr) +} + +func (c *Client) warmupOnce(ctx context.Context, endpoint string) error { + reqURL := c.endpointURL(endpoint, queryOptions{precision: true}) + req, err := http.NewRequestWithContext(ctx, http.MethodGet, reqURL.String(), nil) + if err != nil { + return fmt.Errorf("create request for %s: %w", endpoint, err) + } + + resp, err := c.httpClient.Do(req) + if err != nil { + return fmt.Errorf("fetch %s: %w", endpoint, err) + } + defer resp.Body.Close() + + body, err := io.ReadAll(io.LimitReader(resp.Body, 10<<20)) + if err != nil { + return fmt.Errorf("read %s response: %w", endpoint, err) + } + if resp.StatusCode < 200 || resp.StatusCode >= 300 { + return fmt.Errorf("fetch %s: unexpected HTTP status %d: %s", endpoint, resp.StatusCode, strings.TrimSpace(string(body))) + } + return nil +} + +func (c *Client) fetchHTTP(ctx context.Context, endpoint string, opts queryOptions) (*url.URL, []byte, error) { + attempts := positiveAttemptCount(c.fetchAttempts) + var lastErr error + var lastRetryable bool + for attempt := 1; attempt <= attempts; attempt++ { + if err := ctx.Err(); err != nil { + return nil, nil, fmt.Errorf("fetch %s: %w", endpoint, err) + } + reqURL, body, err := c.fetchHTTPOnce(ctx, endpoint, opts) + if err == nil { + return reqURL, body, nil + } + lastErr = err + lastRetryable = isRetryableRequestError(err) + if !lastRetryable || attempt == attempts { + break + } + if err := waitForRetry(ctx, c.fetchRetryDelay); err != nil { + return nil, nil, fmt.Errorf("fetch %s retry delay after attempt %d: %w", endpoint, attempt, err) + } + } + if lastRetryable { + return nil, nil, fmt.Errorf("fetch %s failed after %d attempts: %w", endpoint, attempts, lastErr) + } + return nil, nil, lastErr +} + +func (c *Client) fetchHTTPOnce(ctx context.Context, endpoint string, opts queryOptions) (*url.URL, []byte, error) { + reqURL := c.endpointURL(endpoint, opts) + req, err := http.NewRequestWithContext(ctx, http.MethodGet, reqURL.String(), nil) + if err != nil { + return nil, nil, fmt.Errorf("create request for %s: %w", endpoint, err) + } + + resp, err := c.httpClient.Do(req) + if err != nil { + err = fmt.Errorf("fetch %s: %w", endpoint, err) + if ctx.Err() != nil { + return reqURL, nil, err + } + return reqURL, nil, retryableRequestError{err: err} + } + defer resp.Body.Close() + + body, err := io.ReadAll(io.LimitReader(resp.Body, 10<<20)) + if err != nil { + err = fmt.Errorf("read %s response: %w", endpoint, err) + if ctx.Err() != nil { + return reqURL, nil, err + } + return reqURL, nil, retryableRequestError{err: err} + } + if resp.StatusCode < 200 || resp.StatusCode >= 300 { + err := fmt.Errorf("fetch %s: unexpected HTTP status %d: %s", endpoint, resp.StatusCode, strings.TrimSpace(string(body))) + if isRetryableHTTPStatus(resp.StatusCode) { + return reqURL, nil, retryableRequestError{err: err} + } + return reqURL, nil, err + } + return reqURL, body, nil +} + +type retryableRequestError struct { + err error +} + +func (e retryableRequestError) Error() string { + return e.err.Error() +} + +func (e retryableRequestError) Unwrap() error { + return e.err +} + +func isRetryableRequestError(err error) bool { + _, ok := err.(retryableRequestError) + return ok +} + +func isRetryableHTTPStatus(status int) bool { + switch status { + case http.StatusRequestTimeout, + http.StatusTooManyRequests, + http.StatusInternalServerError, + http.StatusBadGateway, + http.StatusServiceUnavailable, + http.StatusGatewayTimeout: + return true + default: + return false + } +} + +func waitForRetry(ctx context.Context, delay time.Duration) error { + if delay <= 0 { + return ctx.Err() + } + timer := time.NewTimer(delay) + defer timer.Stop() + select { + case <-ctx.Done(): + return ctx.Err() + case <-timer.C: + return nil + } +} + +func positiveAttemptCount(attempts int) int { + if attempts < 1 { + return 1 + } + return attempts +} + func isJSONNull(raw json.RawMessage) bool { return bytes.Equal(bytes.TrimSpace(raw), []byte("null")) } diff --git a/internal/adapters/weatherapi/client_test.go b/internal/adapters/weatherapi/client_test.go index a809cbe..405a9fe 100644 --- a/internal/adapters/weatherapi/client_test.go +++ b/internal/adapters/weatherapi/client_test.go @@ -80,8 +80,11 @@ func TestFetchBundleFromFixtures(t *testing.T) { "/weatherstories/latest", convectiveOutlooksEndpoint, } - if len(requested) != len(wantPaths) { - t.Fatalf("requested paths = %v, want %d source endpoints", requested, len(wantPaths)) + if len(requested) != len(wantPaths)+1 { + t.Fatalf("requested paths = %v, want warmup plus %d source endpoints", requested, len(wantPaths)) + } + if !strings.HasPrefix(requested[0], defaultWarmupEndpoint+"?") && requested[0] != defaultWarmupEndpoint { + t.Fatalf("first requested path = %q, want warmup endpoint %s", requested[0], defaultWarmupEndpoint) } for _, want := range wantPaths { if !containsPath(requested, want) { @@ -186,7 +189,7 @@ func TestFetchBundleRecordsSourceHash(t *testing.T) { func TestHTTPErrorIsActionable(t *testing.T) { server := fixtureServer(t, map[string]handlerOverride{ - "/conditions/current": {status: http.StatusBadGateway, body: `upstream failed`}, + "/forecast/hourly": {status: http.StatusBadGateway, body: `upstream failed`}, }, nil) client := newTestClient(t, server.URL+"/", nil) @@ -194,15 +197,140 @@ func TestHTTPErrorIsActionable(t *testing.T) { if err == nil { t.Fatal("FetchBundle() error = nil, want HTTP error") } - if !strings.Contains(err.Error(), "/conditions/current") || !strings.Contains(err.Error(), "502") { + if !strings.Contains(err.Error(), "/forecast/hourly") || !strings.Contains(err.Error(), "502") { t.Fatalf("error = %q, want endpoint and status", err.Error()) } } +func TestWarmupRetriesBeforeFetchBundle(t *testing.T) { + var requested []string + var warmupCalls int + server := fixtureServer(t, map[string]handlerOverride{ + defaultWarmupEndpoint: {handler: func(w http.ResponseWriter, r *http.Request) { + warmupCalls++ + if warmupCalls == 1 { + w.WriteHeader(http.StatusBadGateway) + _, _ = w.Write([]byte("vpn waking up")) + return + } + http.ServeFile(w, r, filepath.Join("testdata", "current.json")) + }}, + }, &requested) + client := newTestClient(t, server.URL+"/", nil) + + bundle, err := client.FetchBundle(context.Background()) + if err != nil { + t.Fatalf("FetchBundle() error = %v", err) + } + if bundle.Current == nil { + t.Fatal("Current = nil, want successful fetch after warmup retry") + } + if warmupCalls != 3 { + t.Fatalf("conditions/current calls = %d, want failed warmup, successful warmup, and current source fetch", warmupCalls) + } + if len(requested) < 2 || !containsPath(requested[:2], defaultWarmupEndpoint) { + t.Fatalf("initial requests = %v, want warmup endpoint retries", requested) + } +} + +func TestWarmupFailureStopsBeforeSourceFetches(t *testing.T) { + var requested []string + server := fixtureServer(t, map[string]handlerOverride{ + defaultWarmupEndpoint: {status: http.StatusBadGateway, body: `vpn unavailable`}, + }, &requested) + client := newTestClient(t, server.URL+"/", nil) + client.warmupAttempts = 2 + + _, err := client.FetchBundle(context.Background()) + if err == nil { + t.Fatal("FetchBundle() error = nil, want warmup failure") + } + if !strings.Contains(err.Error(), "warm up weather API") || + !strings.Contains(err.Error(), defaultWarmupEndpoint) || + !strings.Contains(err.Error(), "2 attempts") || + !strings.Contains(err.Error(), "502") { + t.Fatalf("error = %q, want warmup endpoint, attempts, and status", err.Error()) + } + if got := countPath(requested, defaultWarmupEndpoint); got != 2 { + t.Fatalf("warmup requests = %d, want 2; all requests = %v", got, requested) + } + if containsPath(requested, "/observations") { + t.Fatalf("requested paths = %v, want warmup failure before source fetches", requested) + } +} + +func TestFetchRetriesRetryableStatus(t *testing.T) { + var hourlyCalls int + server := fixtureServer(t, map[string]handlerOverride{ + "/forecast/hourly": {handler: func(w http.ResponseWriter, r *http.Request) { + hourlyCalls++ + if hourlyCalls == 1 { + w.WriteHeader(http.StatusBadGateway) + _, _ = w.Write([]byte("temporary upstream failure")) + return + } + http.ServeFile(w, r, filepath.Join("testdata", "hourly.json")) + }}, + }, nil) + client := newTestClient(t, server.URL+"/", nil) + + bundle, err := client.FetchBundle(context.Background()) + if err != nil { + t.Fatalf("FetchBundle() error = %v", err) + } + if bundle.Hourly == nil { + t.Fatal("Hourly = nil, want successful fetch after retry") + } + if hourlyCalls != 2 { + t.Fatalf("hourly calls = %d, want 2", hourlyCalls) + } +} + +func TestFetchDoesNotRetryNonRetryableStatus(t *testing.T) { + var hourlyCalls int + server := fixtureServer(t, map[string]handlerOverride{ + "/forecast/hourly": {handler: func(w http.ResponseWriter, r *http.Request) { + hourlyCalls++ + w.WriteHeader(http.StatusNotFound) + _, _ = w.Write([]byte("not found")) + }}, + }, nil) + client := newTestClient(t, server.URL+"/", nil) + + _, err := client.FetchBundle(context.Background()) + if err == nil { + t.Fatal("FetchBundle() error = nil, want non-retryable status error") + } + if hourlyCalls != 1 { + t.Fatalf("hourly calls = %d, want no retry", hourlyCalls) + } +} + +func TestFetchDoesNotRetryMalformedEnvelope(t *testing.T) { + var hourlyCalls int + server := fixtureServer(t, map[string]handlerOverride{ + "/forecast/hourly": {handler: func(w http.ResponseWriter, r *http.Request) { + hourlyCalls++ + w.WriteHeader(http.StatusOK) + _, _ = w.Write([]byte(`not-json`)) + }}, + }, nil) + client := newTestClient(t, server.URL+"/", nil) + + _, err := client.FetchBundle(context.Background()) + if err == nil { + t.Fatal("FetchBundle() error = nil, want envelope decode error") + } + if hourlyCalls != 1 { + t.Fatalf("hourly calls = %d, want no retry", hourlyCalls) + } +} + func TestRequiredHourlyForecast(t *testing.T) { + var requested []string server := fixtureServer(t, map[string]handlerOverride{ "/forecast/hourly": {status: http.StatusOK, body: `{"data": null}`}, - }, nil) + }, &requested) client := newTestClient(t, server.URL+"/", nil) _, err := client.FetchBundle(context.Background()) @@ -212,6 +340,9 @@ func TestRequiredHourlyForecast(t *testing.T) { if !strings.Contains(err.Error(), "hourly forecast data") { t.Fatalf("error = %q, want hourly context", err.Error()) } + if got := countPath(requested, "/forecast/hourly"); got != 1 { + t.Fatalf("hourly requests = %d, want no retry; all requests = %v", got, requested) + } } func TestNullAlertsMeansNoActiveAlerts(t *testing.T) { @@ -420,6 +551,43 @@ func TestContextCancellation(t *testing.T) { } } +func TestRetryDelayRespectsContextCancellation(t *testing.T) { + var cancel context.CancelFunc + var hourlyCalls int + server := fixtureServer(t, map[string]handlerOverride{ + "/forecast/hourly": {handler: func(w http.ResponseWriter, r *http.Request) { + hourlyCalls++ + if cancel != nil { + cancel() + } + w.WriteHeader(http.StatusBadGateway) + _, _ = w.Write([]byte("temporary upstream failure")) + }}, + }, nil) + client := newTestClient(t, server.URL+"/", nil) + client.fetchRetryDelay = time.Hour + + ctx, cancelFunc := context.WithCancel(context.Background()) + cancel = cancelFunc + defer cancelFunc() + + start := time.Now() + _, err := client.FetchBundle(ctx) + elapsed := time.Since(start) + if err == nil { + t.Fatal("FetchBundle() error = nil, want cancellation during retry delay") + } + if !strings.Contains(err.Error(), context.Canceled.Error()) { + t.Fatalf("error = %q, want context cancellation", err.Error()) + } + if elapsed > time.Second { + t.Fatalf("FetchBundle() elapsed = %s, want prompt cancellation", elapsed) + } + if hourlyCalls != 1 { + t.Fatalf("hourly calls = %d, want retry delay cancellation before second attempt", hourlyCalls) + } +} + func TestHTTPTimeout(t *testing.T) { server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { time.Sleep(50 * time.Millisecond) @@ -432,12 +600,14 @@ func TestHTTPTimeout(t *testing.T) { if err != nil { t.Fatalf("New() error = %v", err) } + client.warmupDelay = 0 + client.fetchRetryDelay = 0 _, err = client.FetchBundle(context.Background()) if err == nil { t.Fatal("FetchBundle() error = nil, want timeout error") } - if !strings.Contains(err.Error(), "/observations") { + if !strings.Contains(err.Error(), defaultWarmupEndpoint) { t.Fatalf("error = %q, want endpoint context", err.Error()) } } @@ -464,8 +634,9 @@ func TestSaveBundle(t *testing.T) { } type handlerOverride struct { - status int - body string + status int + body string + handler http.HandlerFunc } func fixtureServer(t *testing.T, overrides map[string]handlerOverride, requested *[]string) *httptest.Server { @@ -485,6 +656,10 @@ func fixtureServer(t *testing.T, overrides map[string]handlerOverride, requested *requested = append(*requested, r.URL.String()) } if override, ok := overrides[r.URL.Path]; ok { + if override.handler != nil { + override.handler(w, r) + return + } w.WriteHeader(override.status) _, _ = w.Write([]byte(override.body)) return @@ -510,6 +685,8 @@ func newTestClient(t *testing.T, baseURL string, sourcePolicies map[string]confi if err != nil { t.Fatalf("New() error = %v", err) } + client.warmupDelay = 0 + client.fetchRetryDelay = 0 return client } @@ -532,6 +709,16 @@ func containsPath(requested []string, path string) bool { return false } +func countPath(requested []string, path string) int { + var count int + for _, rawURL := range requested { + if strings.HasPrefix(rawURL, path+"?") || rawURL == path { + count++ + } + } + return count +} + func sourceByName(t *testing.T, sources []weatherdata.Source, name string) weatherdata.Source { t.Helper() for _, source := range sources {