Fetch independent weather sources concurrently

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, &current)
-	} else {
-		fetched, ok, err = b.fetchDecodedSource(ctx, request, &current)
-	}
+	fetched, ok, err := b.fetchDecodedSource(acquired, &current)
 	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 {

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