From 0516ee148d6d3ed8c7a50e87f11126f9f83116f5 Mon Sep 17 00:00:00 2001 From: Eric Rakestraw Date: Thu, 13 Aug 2026 03:45:55 +0000 Subject: [PATCH] Fetch independent weather sources concurrently --- docs/integrations/weatherapi.md | 5 +- docs/internal/collect.md | 5 + internal/adapters/weatherapi/client.go | 227 ++++++++++---------- internal/adapters/weatherapi/client_test.go | 212 ++++++++++++++++-- 4 files changed, 317 insertions(+), 132 deletions(-) diff --git a/docs/integrations/weatherapi.md b/docs/integrations/weatherapi.md index 2098c9b..0e9972d 100644 --- a/docs/integrations/weatherapi.md +++ b/docs/integrations/weatherapi.md @@ -28,7 +28,10 @@ fetch before source requests begin. The adapter makes one source request for each endpoint, subject to retry on transient failures. A successful readiness request supplies the current -conditions source response. +conditions source response. The remaining independent source requests run +concurrently, then their results are processed in the source order shown below. +This keeps source provenance, missing-source policy, and surfaced errors +deterministic regardless of response order. | Source | Endpoint | Query parameters | Availability | | --- | --- | --- | --- | diff --git a/docs/internal/collect.md b/docs/internal/collect.md index 71b5798..00249b5 100644 --- a/docs/internal/collect.md +++ b/docs/internal/collect.md @@ -15,6 +15,11 @@ The adapter's successful readiness request for current conditions is reused as that normalized source; collection does not trigger a second identical current conditions request. +After readiness succeeds, the adapter obtains the other independent source +responses concurrently. It merges their normalized results in the established +source order, so provenance, warnings, and source-local failures remain +deterministic. Cancellation remains authoritative for every in-flight request. + The package wraps adapter construction failures, including an invalid Weather API base URL, as weather-collection setup errors and fetch failures as bundle-collection errors. It does not retry, persist, select reports, derive diff --git a/internal/adapters/weatherapi/client.go b/internal/adapters/weatherapi/client.go index be540e9..7540c0f 100644 --- a/internal/adapters/weatherapi/client.go +++ b/internal/adapters/weatherapi/client.go @@ -15,6 +15,7 @@ import ( "path" "strconv" "strings" + "sync" "time" "gitea.maximumdirect.net/eric/weatherreporter/internal/config" @@ -121,43 +122,25 @@ func (c *Client) FetchBundle(ctx context.Context) (*weatherdata.Bundle, error) { fetchedAt := c.now() builder := bundleBuilder{ - client: c, - bundle: &weatherdata.Bundle{FetchedAt: fetchedAt}, - fetchedAt: fetchedAt, + client: c, + bundle: &weatherdata.Bundle{FetchedAt: fetchedAt}, } - if err := builder.fetchObservation(ctx); err != nil { - return nil, err - } - if err := builder.fetchCurrent(ctx, warmup); err != nil { - return nil, err - } - if err := builder.fetchHourly(ctx); err != nil { - return nil, err - } - if err := builder.fetchNarrative(ctx); err != nil { - return nil, err - } - if err := builder.fetchAlerts(ctx); err != nil { - return nil, err - } - if err := builder.fetchDiscussion(ctx); err != nil { - return nil, err - } - if err := builder.fetchWeatherStory(ctx); err != nil { - return nil, err - } - if err := builder.fetchSPCConvectiveOutlooks(ctx); err != nil { - return nil, err + for _, acquired := range builder.acquireSources(ctx, warmup) { + if err := ctx.Err(); err != nil { + return nil, fmt.Errorf("fetch weather API sources: %w", err) + } + if err := builder.mergeSource(acquired); err != nil { + return nil, err + } } return builder.bundle, nil } type bundleBuilder struct { - client *Client - bundle *weatherdata.Bundle - fetchedAt time.Time + client *Client + bundle *weatherdata.Bundle } type sourceRequest struct { @@ -181,14 +164,74 @@ type warmupResponse struct { fetchedAt time.Time } -func (b *bundleBuilder) fetchObservation(ctx context.Context) error { +type sourceAcquisition struct { + request sourceRequest + fetched fetchedSource + err error + warmup warmupResponse + usesWarmup bool +} + +func (b *bundleBuilder) acquireSources(ctx context.Context, warmup warmupResponse) []sourceAcquisition { + sources := []sourceAcquisition{ + {request: sourceRequest{name: config.MissingSourceObservations, endpoint: "/observations", query: queryOptions{precision: true}, missingMessage: "observation data is missing"}}, + {request: currentConditionsRequest()}, + {request: sourceRequest{name: "hourly", endpoint: "/forecast/hourly", query: queryOptions{precision: true, timezone: true}, missingMessage: "hourly forecast data is missing", required: true, decodeLabel: "hourly forecast"}}, + {request: sourceRequest{name: config.MissingSourceNarrative, endpoint: "/forecast/narrative", query: queryOptions{precision: true, timezone: true}, missingMessage: "narrative forecast data is missing"}}, + {request: sourceRequest{name: config.MissingSourceAlerts, endpoint: "/alerts/active", query: queryOptions{allowNull: true}, missingMessage: "active alerts data is missing"}}, + {request: sourceRequest{name: config.MissingSourceDiscussion, endpoint: "/discussion", query: queryOptions{timezone: true}, missingMessage: "forecast discussion data is missing"}}, + {request: sourceRequest{name: config.MissingSourceWeatherStory, endpoint: "/weatherstories/latest", query: queryOptions{omitUnits: true}, missingMessage: "NWS weather story data is missing"}}, + {request: sourceRequest{name: sourceSPCConvectiveOutlooks, endpoint: convectiveOutlooksEndpoint, query: queryOptions{timezone: true, omitUnits: true}, missingMessage: "SPC convective outlook data is missing"}}, + } + if warmup.endpoint == currentConditionsEndpoint { + sources[1].warmup = warmup + sources[1].usesWarmup = true + } + + var group sync.WaitGroup + for i := range sources { + if sources[i].usesWarmup { + continue + } + group.Add(1) + go func(index int) { + defer group.Done() + request := sources[index].request + raw, source, err := b.client.fetch(ctx, request.name, request.endpoint, request.query) + sources[index].fetched = fetchedSource{raw: raw, source: source} + sources[index].err = err + }(i) + } + group.Wait() + return sources +} + +func (b *bundleBuilder) mergeSource(acquired sourceAcquisition) error { + switch acquired.request.name { + case config.MissingSourceObservations: + return b.fetchObservation(acquired) + case config.MissingSourceCurrent: + return b.fetchCurrent(acquired) + case "hourly": + return b.fetchHourly(acquired) + case config.MissingSourceNarrative: + return b.fetchNarrative(acquired) + case config.MissingSourceAlerts: + return b.fetchAlerts(acquired) + case config.MissingSourceDiscussion: + return b.fetchDiscussion(acquired) + case config.MissingSourceWeatherStory: + return b.fetchWeatherStory(acquired) + case sourceSPCConvectiveOutlooks: + return b.fetchSPCConvectiveOutlooks(acquired) + default: + return fmt.Errorf("merge unknown weather source %q", acquired.request.name) + } +} + +func (b *bundleBuilder) fetchObservation(acquired sourceAcquisition) error { var observation weatherdata.Observation - fetched, ok, err := b.fetchDecodedSource(ctx, sourceRequest{ - name: config.MissingSourceObservations, - endpoint: "/observations", - query: queryOptions{precision: true}, - missingMessage: "observation data is missing", - }, &observation) + fetched, ok, err := b.fetchDecodedSource(acquired, &observation) if err != nil || !ok { return err } @@ -208,19 +251,9 @@ func currentConditionsRequest() sourceRequest { } } -func (b *bundleBuilder) fetchCurrent(ctx context.Context, warmup warmupResponse) error { +func (b *bundleBuilder) fetchCurrent(acquired sourceAcquisition) error { var current weatherdata.Current - request := currentConditionsRequest() - var ( - fetched fetchedSource - ok bool - err error - ) - if warmup.endpoint == request.endpoint { - fetched, ok, err = b.fetchWarmupSource(warmup, request, ¤t) - } else { - fetched, ok, err = b.fetchDecodedSource(ctx, request, ¤t) - } + fetched, ok, err := b.fetchDecodedSource(acquired, ¤t) if err != nil || !ok { return err } @@ -230,16 +263,9 @@ func (b *bundleBuilder) fetchCurrent(ctx context.Context, warmup warmupResponse) return nil } -func (b *bundleBuilder) fetchHourly(ctx context.Context) error { +func (b *bundleBuilder) fetchHourly(acquired sourceAcquisition) error { var hourly weatherdata.ForecastRun - fetched, ok, err := b.fetchDecodedSource(ctx, sourceRequest{ - name: "hourly", - endpoint: "/forecast/hourly", - query: queryOptions{precision: true, timezone: true}, - missingMessage: "hourly forecast data is missing", - required: true, - decodeLabel: "hourly forecast", - }, &hourly) + fetched, ok, err := b.fetchDecodedSource(acquired, &hourly) if err != nil || !ok { return err } @@ -262,14 +288,9 @@ func (b *bundleBuilder) fetchHourly(ctx context.Context) error { return nil } -func (b *bundleBuilder) fetchNarrative(ctx context.Context) error { +func (b *bundleBuilder) fetchNarrative(acquired sourceAcquisition) error { var narrative weatherdata.ForecastRun - fetched, ok, err := b.fetchDecodedSource(ctx, sourceRequest{ - name: config.MissingSourceNarrative, - endpoint: "/forecast/narrative", - query: queryOptions{precision: true, timezone: true}, - missingMessage: "narrative forecast data is missing", - }, &narrative) + fetched, ok, err := b.fetchDecodedSource(acquired, &narrative) if err != nil || !ok { return err } @@ -281,14 +302,12 @@ func (b *bundleBuilder) fetchNarrative(ctx context.Context) error { return nil } -func (b *bundleBuilder) fetchAlerts(ctx context.Context) error { - raw, source, err := b.client.fetch(ctx, config.MissingSourceAlerts, "/alerts/active", queryOptions{allowNull: true}) - if err != nil { +func (b *bundleBuilder) fetchAlerts(acquired sourceAcquisition) error { + fetched, ok, err := b.fetchSource(acquired) + if err != nil || !ok { return err } - if raw == nil { - return b.handleMissing(&source, "active alerts data is missing", false) - } + raw, source := fetched.raw, fetched.source if isJSONNull(raw) { b.bundle.Alerts = &weatherdata.AlertRun{Raw: append(json.RawMessage(nil), raw...)} b.addSource(source) @@ -296,7 +315,7 @@ func (b *bundleBuilder) fetchAlerts(ctx context.Context) error { } var alerts weatherdata.AlertRun if err := decodeSource(raw, &alerts); err != nil { - return b.handleMalformed(&source, err, sourceRequest{name: config.MissingSourceAlerts}) + return b.handleMalformed(&source, err, acquired.request) } alerts.Raw = append(json.RawMessage(nil), raw...) if alerts.AsOf != nil { @@ -307,14 +326,9 @@ func (b *bundleBuilder) fetchAlerts(ctx context.Context) error { return nil } -func (b *bundleBuilder) fetchDiscussion(ctx context.Context) error { +func (b *bundleBuilder) fetchDiscussion(acquired sourceAcquisition) error { var discussion weatherdata.Discussion - fetched, ok, err := b.fetchDecodedSource(ctx, sourceRequest{ - name: config.MissingSourceDiscussion, - endpoint: "/discussion", - query: queryOptions{timezone: true}, - missingMessage: "forecast discussion data is missing", - }, &discussion) + fetched, ok, err := b.fetchDecodedSource(acquired, &discussion) if err != nil || !ok { return err } @@ -326,21 +340,15 @@ func (b *bundleBuilder) fetchDiscussion(ctx context.Context) error { return nil } -func (b *bundleBuilder) fetchWeatherStory(ctx context.Context) error { +func (b *bundleBuilder) fetchWeatherStory(acquired sourceAcquisition) error { var story weatherdata.WeatherStory - request := sourceRequest{ - name: config.MissingSourceWeatherStory, - endpoint: "/weatherstories/latest", - query: queryOptions{omitUnits: true}, - missingMessage: "NWS weather story data is missing", - } - fetched, ok, err := b.fetchDecodedSource(ctx, request, &story) + fetched, ok, err := b.fetchDecodedSource(acquired, &story) if err != nil || !ok { return err } source := fetched.source if !story.HasUsableContent() { - return b.handleMalformed(&source, fmt.Errorf("weather story has no usable content"), request) + return b.handleMalformed(&source, fmt.Errorf("weather story has no usable content"), acquired.request) } if !story.StartTime.IsZero() { source.IssuedAt = &story.StartTime @@ -351,14 +359,9 @@ func (b *bundleBuilder) fetchWeatherStory(ctx context.Context) error { return nil } -func (b *bundleBuilder) fetchSPCConvectiveOutlooks(ctx context.Context) error { +func (b *bundleBuilder) fetchSPCConvectiveOutlooks(acquired sourceAcquisition) error { var run weatherdata.ConvectiveOutlookRun - fetched, ok, err := b.fetchDecodedSource(ctx, sourceRequest{ - name: sourceSPCConvectiveOutlooks, - endpoint: convectiveOutlooksEndpoint, - query: queryOptions{timezone: true, omitUnits: true}, - missingMessage: "SPC convective outlook data is missing", - }, &run) + fetched, ok, err := b.fetchDecodedSource(acquired, &run) if err != nil || !ok { return err } @@ -374,23 +377,12 @@ func (b *bundleBuilder) fetchSPCConvectiveOutlooks(ctx context.Context) error { return nil } -func (b *bundleBuilder) fetchDecodedSource(ctx context.Context, request sourceRequest, target any) (fetchedSource, bool, error) { - fetched, ok, err := b.fetchSource(ctx, request) +func (b *bundleBuilder) fetchDecodedSource(acquired sourceAcquisition, target any) (fetchedSource, bool, error) { + fetched, ok, err := b.fetchSource(acquired) if err != nil || !ok { return fetchedSource{}, false, err } - return b.decodeFetchedSource(fetched, request, target) -} - -func (b *bundleBuilder) fetchWarmupSource(warmup warmupResponse, request sourceRequest, target any) (fetchedSource, bool, error) { - raw, source, err := b.client.decodeSourceResponse(request.name, request.endpoint, request.query, warmup.requestURL, warmup.body, warmup.fetchedAt) - if err != nil { - return fetchedSource{}, false, err - } - if raw == nil { - return fetchedSource{}, false, b.handleMissing(&source, request.missingMessage, request.required) - } - return b.decodeFetchedSource(fetchedSource{raw: raw, source: source}, request, target) + return b.decodeFetchedSource(fetched, acquired.request, target) } func (b *bundleBuilder) decodeFetchedSource(fetched fetchedSource, request sourceRequest, target any) (fetchedSource, bool, error) { @@ -400,15 +392,20 @@ func (b *bundleBuilder) decodeFetchedSource(fetched fetchedSource, request sourc return fetched, true, nil } -func (b *bundleBuilder) fetchSource(ctx context.Context, request sourceRequest) (fetchedSource, bool, error) { - raw, source, err := b.client.fetch(ctx, request.name, request.endpoint, request.query) - if err != nil { - return fetchedSource{}, false, err +func (b *bundleBuilder) fetchSource(acquired sourceAcquisition) (fetchedSource, bool, error) { + if acquired.usesWarmup { + raw, source, err := b.client.decodeSourceResponse(acquired.request.name, acquired.request.endpoint, acquired.request.query, acquired.warmup.requestURL, acquired.warmup.body, acquired.warmup.fetchedAt) + if err != nil { + return fetchedSource{}, false, err + } + acquired.fetched = fetchedSource{raw: raw, source: source} + } else if acquired.err != nil { + return fetchedSource{}, false, acquired.err } - if raw == nil { - return fetchedSource{}, false, b.handleMissing(&source, request.missingMessage, request.required) + if acquired.fetched.raw == nil { + return fetchedSource{}, false, b.handleMissing(&acquired.fetched.source, acquired.request.missingMessage, acquired.request.required) } - return fetchedSource{raw: raw, source: source}, true, nil + return acquired.fetched, true, nil } func (b *bundleBuilder) handleMissing(source *weatherdata.Source, message string, required bool) error { diff --git a/internal/adapters/weatherapi/client_test.go b/internal/adapters/weatherapi/client_test.go index 3d224b1..eeea722 100644 --- a/internal/adapters/weatherapi/client_test.go +++ b/internal/adapters/weatherapi/client_test.go @@ -9,6 +9,7 @@ import ( "os" "path/filepath" "strings" + "sync" "testing" "time" @@ -115,6 +116,175 @@ func TestFetchBundleFromFixtures(t *testing.T) { } } +func TestFetchBundleMergesConcurrentSourcesInSourceOrder(t *testing.T) { + paths := []string{ + "/observations", + "/forecast/hourly", + "/forecast/narrative", + "/alerts/active", + "/discussion", + "/weatherstories/latest", + convectiveOutlooksEndpoint, + } + started := make(chan string, len(paths)) + release := make(map[string]chan struct{}, len(paths)) + for _, path := range paths { + release[path] = make(chan struct{}) + } + var releaseOnce sync.Once + releaseAll := func() { + releaseOnce.Do(func() { + for i := len(paths) - 1; i >= 0; i-- { + close(release[paths[i]]) + } + }) + } + t.Cleanup(releaseAll) + server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + if r.URL.Path == currentConditionsEndpoint { + if !serveWeatherFixture(w, r) { + http.NotFound(w, r) + } + return + } + ready, ok := release[r.URL.Path] + if !ok { + http.NotFound(w, r) + return + } + started <- r.URL.Path + <-ready + if !serveWeatherFixture(w, r) { + http.NotFound(w, r) + } + })) + defer server.Close() + client := newTestClient(t, server.URL+"/", nil) + + type fetchResult struct { + bundle *weatherdata.Bundle + err error + } + result := make(chan fetchResult, 1) + go func() { + bundle, err := client.FetchBundle(context.Background()) + result <- fetchResult{bundle: bundle, err: err} + }() + + seen := make(map[string]bool, len(paths)) + for range paths { + select { + case path := <-started: + seen[path] = true + case <-time.After(time.Second): + t.Fatalf("independent requests started = %v, want %v", seen, paths) + } + } + releaseAll() + + select { + case got := <-result: + if got.err != nil { + t.Fatalf("FetchBundle() error = %v", got.err) + } + wantSources := []string{ + config.MissingSourceObservations, + config.MissingSourceCurrent, + "hourly", + config.MissingSourceNarrative, + config.MissingSourceAlerts, + config.MissingSourceDiscussion, + config.MissingSourceWeatherStory, + sourceSPCConvectiveOutlooks, + } + gotSources := make([]string, 0, len(got.bundle.Sources)) + for _, source := range got.bundle.Sources { + gotSources = append(gotSources, source.Name) + } + if strings.Join(gotSources, ",") != strings.Join(wantSources, ",") { + t.Fatalf("source order = %v, want %v", gotSources, wantSources) + } + case <-time.After(time.Second): + t.Fatal("FetchBundle() did not finish after all source responses were released") + } +} + +func TestFetchBundleReportsConcurrentFailuresInSourceOrder(t *testing.T) { + var requested []string + server := fixtureServer(t, map[string]handlerOverride{ + "/forecast/hourly": {status: http.StatusBadRequest, body: `invalid hourly request`}, + "/forecast/narrative": {status: http.StatusBadRequest, body: `invalid narrative request`}, + }, &requested) + client := newTestClient(t, server.URL+"/", nil) + + _, err := client.FetchBundle(context.Background()) + if err == nil { + t.Fatal("FetchBundle() error = nil, want source error") + } + if !strings.Contains(err.Error(), "/forecast/hourly") { + t.Fatalf("error = %q, want the earlier hourly source failure", err.Error()) + } + if !containsPath(requested, "/forecast/narrative") { + t.Fatalf("requested paths = %v, want independent narrative request", requested) + } +} + +func TestFetchBundleCancelsConcurrentSourceRequests(t *testing.T) { + paths := []string{ + "/observations", + "/forecast/hourly", + "/forecast/narrative", + "/alerts/active", + "/discussion", + "/weatherstories/latest", + convectiveOutlooksEndpoint, + } + started := make(chan string, len(paths)) + server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + if r.URL.Path == currentConditionsEndpoint { + if !serveWeatherFixture(w, r) { + http.NotFound(w, r) + } + return + } + for _, path := range paths { + if r.URL.Path == path { + started <- path + <-r.Context().Done() + return + } + } + http.NotFound(w, r) + })) + defer server.Close() + client := newTestClient(t, server.URL+"/", nil) + ctx, cancel := context.WithCancel(context.Background()) + defer cancel() + + result := make(chan error, 1) + go func() { + _, err := client.FetchBundle(ctx) + result <- err + }() + for range paths { + select { + case <-started: + case <-time.After(time.Second): + cancel() + t.Fatal("not all independent requests started before cancellation") + } + } + cancel() + select { + case err := <-result: + if err == nil || !strings.Contains(err.Error(), context.Canceled.Error()) { + t.Fatalf("FetchBundle() error = %v, want context cancellation", err) + } + case <-time.After(time.Second): + t.Fatal("FetchBundle() did not return after cancellation") + } +} + func TestFetchBundleRejectsInvalidHourlyPrecipitationProbability(t *testing.T) { for _, probability := range []string{"-1", "101"} { t.Run(probability, func(t *testing.T) { @@ -369,8 +539,8 @@ func TestFetchRejectsOversizedResponseWithoutRetry(t *testing.T) { if narrativeCalls != 1 { t.Fatalf("narrative calls = %d, want no retry", narrativeCalls) } - if containsPath(requested, "/alerts/active") { - t.Fatalf("requested paths = %v, want source failure before later fetches", requested) + if !containsPath(requested, "/alerts/active") { + t.Fatalf("requested paths = %v, want independent source requests despite narrative failure", requested) } } @@ -912,21 +1082,34 @@ type handlerOverride struct { handler http.HandlerFunc } +var weatherFixtureFiles = map[string]string{ + "/observations": "observations.json", + "/conditions/current": "current.json", + "/forecast/hourly": "hourly.json", + "/forecast/narrative": "narrative.json", + "/alerts/active": "alerts.json", + "/discussion": "discussion.json", + "/weatherstories/latest": "weather_story.json", + convectiveOutlooksEndpoint: "convective_outlooks.json", +} + +func serveWeatherFixture(w http.ResponseWriter, r *http.Request) bool { + name, ok := weatherFixtureFiles[r.URL.Path] + if !ok { + return false + } + http.ServeFile(w, r, filepath.Join("testdata", name)) + return true +} + func fixtureServer(t *testing.T, overrides map[string]handlerOverride, requested *[]string) *httptest.Server { t.Helper() - fixtures := map[string]string{ - "/observations": "observations.json", - "/conditions/current": "current.json", - "/forecast/hourly": "hourly.json", - "/forecast/narrative": "narrative.json", - "/alerts/active": "alerts.json", - "/discussion": "discussion.json", - "/weatherstories/latest": "weather_story.json", - convectiveOutlooksEndpoint: "convective_outlooks.json", - } + var requestedMu sync.Mutex server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { if requested != nil { + requestedMu.Lock() *requested = append(*requested, r.URL.String()) + requestedMu.Unlock() } if override, ok := overrides[r.URL.Path]; ok { if override.handler != nil { @@ -937,12 +1120,9 @@ func fixtureServer(t *testing.T, overrides map[string]handlerOverride, requested _, _ = w.Write([]byte(override.body)) return } - name, ok := fixtures[r.URL.Path] - if !ok { + if !serveWeatherFixture(w, r) { http.NotFound(w, r) - return } - http.ServeFile(w, r, filepath.Join("testdata", name)) })) t.Cleanup(server.Close) return server