From 84d1664b2967094c68e65ebe104b486b325175b9 Mon Sep 17 00:00:00 2001 From: gili8420 Date: Tue, 4 Aug 2026 13:40:50 +0900 Subject: [PATCH] fix(input): correct lon mapping --- .gitignore | 1 + api/rest/predictor.swagger.yml | 7 + cmd/compare-tawhiri/main.go | 220 ++++++++++++++++++++++++-- internal/api/mapping.go | 62 +++++++- internal/api/mapping_test.go | 88 +++++++++++ internal/api/prediction.go | 90 ++++++++--- internal/api/transport.go | 6 +- internal/datasets/manager.go | 101 +++++++++--- internal/datasets/manager_test.go | 166 +++++++++++++++++++ internal/engine/engine_test.go | 156 ++++++++++++++++++ internal/engine/events.go | 41 ++++- internal/engine/models.go | 7 + internal/engine/registry.go | 52 ++++-- internal/windviz/windviz.go | 29 +++- internal/windviz/windviz_test.go | 85 ++++++++++ pkg/rest/oas_client_gen.go | 121 ++++++++++++-- pkg/rest/oas_handlers_gen.go | 30 ++-- pkg/rest/oas_json_gen.go | 31 +++- pkg/rest/oas_response_encoders_gen.go | 21 +-- pkg/rest/oas_schemas_gen.go | 20 ++- 20 files changed, 1197 insertions(+), 137 deletions(-) create mode 100644 .gitignore create mode 100644 internal/api/mapping_test.go create mode 100644 internal/datasets/manager_test.go diff --git a/.gitignore b/.gitignore new file mode 100644 index 0000000..ba077a4 --- /dev/null +++ b/.gitignore @@ -0,0 +1 @@ +bin diff --git a/api/rest/predictor.swagger.yml b/api/rest/predictor.swagger.yml index 67534c7..b959722 100644 --- a/api/rest/predictor.swagger.yml +++ b/api/rest/predictor.swagger.yml @@ -375,6 +375,13 @@ components: profile, hand off to `fallback_index`, or clip to the boundary. properties: launch: { $ref: "#/components/schemas/Launch" } + dataset: + type: string + format: date-time + description: | + Forecast run to predict from, given as its epoch. Defaults to the + active dataset. The named run must be stored; a request for one that + is not is rejected rather than answered from a different run. direction: type: string enum: [forward, reverse] diff --git a/cmd/compare-tawhiri/main.go b/cmd/compare-tawhiri/main.go index c6ea5ee..75fcd13 100644 --- a/cmd/compare-tawhiri/main.go +++ b/cmd/compare-tawhiri/main.go @@ -24,6 +24,7 @@ import ( "net/http" "net/url" "os" + "strings" "text/tabwriter" "time" ) @@ -75,15 +76,22 @@ func main() { var worst float64 compared := 0 for _, s := range sites { - p := params{lat: s.lat, lng: s.lng, alt: *alt, launch: launchTime, + // The site suite compares trajectories, not notations, so fold western + // longitudes into [0, 360) here: that is the only range Tawhiri accepts. + // The longitude suite below sends raw values precisely to test that edge. + lng := s.lng + if lng < 0 { + lng += 360 + } + p := params{lat: s.lat, lng: lng, alt: *alt, launch: launchTime, ascent: *ascent, burst: *burst, descent: *descent} - ours, err := predict(*server+"/api/v1/prediction", p, "") + ours, _, err := predict(*server+"/api/v1/prediction", p, "") if err != nil { fmt.Fprintf(tw, "%s\tlocal error: %v\n", s.name, err) continue } - theirs, err := predict(*tawhiri, p, datasetParam) + theirs, _, err := predict(*tawhiri, p, datasetParam) if err != nil { fmt.Fprintf(tw, "%s\ttawhiri error: %v\n", s.name, err) continue @@ -106,10 +114,17 @@ func main() { } tw.Flush() + breaks := comparePairs(*server, *tawhiri, longitudeParams(launchTime, *alt, *ascent, *burst, *descent), datasetParam) + breaks += compareLongitudes(*server, *tawhiri, longitudeParams(launchTime, *alt, *ascent, *burst, *descent), datasetParam) + if compared == 0 { fmt.Println("\nVERDICT: NO COMPARISONS (every site errored — see rows above)") os.Exit(1) } + if breaks > 0 { + fmt.Printf("\nVERDICT: INCOMPATIBLE (%d longitude(s) Tawhiri accepts and we reject)\n", breaks) + os.Exit(3) + } fmt.Printf("\ncompared %d/%d sites; worst landing distance: %.2f km\n", compared, len(sites), worst/1000) switch { case worst < 1000: @@ -153,17 +168,20 @@ type result struct { dataset string } -func predict(endpoint string, p params, dataset string) (result, error) { - // Tawhiri requires longitude in [0, 360); normalize so both endpoints get - // the same request. Returned trajectory longitudes are [-180, 180] on both - // sides, so the comparison stays consistent. - lng := p.lng - if lng < 0 { - lng += 360 - } +// predict sends p verbatim and returns the parsed result plus the HTTP status. +// +// Nothing is normalised here on purpose. This function used to fold negative +// longitudes into [0, 360) before sending, which made every longitude notation +// look identical to both endpoints — so the tool could not see a disagreement +// about notation even in principle, and ours drifted to [-180, 360) with an +// asymmetry nobody had chosen. Callers that want a fold do it themselves. +// +// The status is returned separately from the error so a caller can tell a +// deliberate rejection from a transport failure. +func predict(endpoint string, p params, dataset string) (result, int, error) { q := url.Values{} q.Set("launch_latitude", fmt.Sprintf("%.4f", p.lat)) - q.Set("launch_longitude", fmt.Sprintf("%.4f", lng)) + q.Set("launch_longitude", fmt.Sprintf("%.4f", p.lng)) q.Set("launch_altitude", fmt.Sprintf("%.0f", p.alt)) q.Set("launch_datetime", p.launch.Format(time.RFC3339)) q.Set("ascent_rate", fmt.Sprintf("%.2f", p.ascent)) @@ -191,10 +209,10 @@ func predict(endpoint string, p params, dataset string) (result, error) { break } if lastErr != nil { - return result{}, lastErr + return result{}, 0, lastErr } if status != 200 { - return result{}, fmt.Errorf("HTTP %d: %s", status, truncate(string(body), 160)) + return result{}, status, fmt.Errorf("HTTP %d: %s", status, truncate(string(body), 160)) } var doc struct { @@ -211,7 +229,7 @@ func predict(endpoint string, p params, dataset string) (result, error) { } `json:"request"` } if err := json.Unmarshal(body, &doc); err != nil { - return result{}, err + return result{}, status, err } var r result @@ -230,7 +248,7 @@ func predict(endpoint string, p params, dataset string) (result, error) { r.landLat, r.landLng, r.landAlt = last.Latitude, last.Longitude, last.Altitude } } - return r, nil + return r, status, nil } type readinessResp struct { @@ -274,3 +292,173 @@ func truncate(s string, n int) string { } return s[:n] + "…" } + +// longitudeCases are the notations a client can put in launch_longitude, sent +// verbatim to both endpoints. +// +// Longitude is on a circle, so every one of these names a real meridian and no +// value here is geometrically wrong; what differs is which notations each service +// agrees to read. Tawhiri implements a half-open [0, 360) — one clean fundamental +// domain, every meridian spelled exactly once. We accept a superset deliberately, +// because the frontend holds longitudes in [-180, 180] and a western launch +// therefore arrives negative. +// +// The pairs below name the same meridian in the two notations, so an accepted +// pair must produce the same trajectory: (-51.7, 308.3), (-180, 180), (0, 360), +// (-90, 270). That is what makes this a comparison rather than a status table. +func longitudeCases() []float64 { + return []float64{ + 0, 180, 270, 308.3, 359.999, // inside Tawhiri's domain + -0.0001, -51.7, -90, -180, -200, // signed convention: our extension + 360, // 0 spelled redundantly; Tawhiri refuses it + 5170, // a plain typo, refused by both + } +} + +func longitudeParams(launch time.Time, alt, ascent, burst, descent float64) params { + // 64.1 N is Nuuk's latitude: high enough to be a realistic Arctic site, far + // enough from the pole that the comparison is about longitude alone. + return params{lat: 64.1, alt: alt, launch: launch, ascent: ascent, burst: burst, descent: descent} +} + +// compareLongitudes reports how the two services read each notation and returns +// the number of compatibility breaks — longitudes Tawhiri accepts and we refuse. +// Those are the only rows that are a defect: refusing what a Tawhiri client is +// entitled to send is what a drop-in replacement must never do. The reverse, us +// accepting what Tawhiri refuses, is the extension the product depends on. +func compareLongitudes(server, tawhiri string, base params, datasetParam string) int { + tw := tabwriter.NewWriter(os.Stdout, 0, 0, 2, ' ', 0) + fmt.Fprintln(tw, "\nlongitude\tours\ttawhiri\tlanding Δ\tverdict") + fmt.Fprintln(tw, "---------\t----\t-------\t---------\t-------") + + breaks := 0 + for _, lng := range longitudeCases() { + p := base + p.lng = lng + + ours, oursStatus, oursErr := predict(server+"/api/v1/prediction", p, "") + theirs, theirsStatus, theirsErr := predict(tawhiri, p, datasetParam) + + code := func(status int, err error) string { + if err == nil { + return "200" + } + if status == 0 { + return "transport" + } + return fmt.Sprintf("%d", status) + } + + delta := "" + verdict := "" + switch { + case oursErr == nil && theirsErr == nil: + d := haversine(ours.landLat, ours.landLng, theirs.landLat, theirs.landLng) + delta = fmt.Sprintf("%.2f km", d/1000) + verdict = "agree" + if d > 50000 { + verdict = "ACCEPTED BY BOTH, TRAJECTORIES DIVERGE" + } + // v1 must publish the convention Tawhiri publishes, not merely the same + // place in a different notation. It used to answer -51.2115 where + // Tawhiri answers 308.7884; haversine cannot see that, because a 360 + // degree difference vanishes from sin(dlam/2). + if (ours.landLng < 0) != (theirs.landLng < 0) { + verdict = fmt.Sprintf("OUTPUT FORMAT DIFFERS (ours %.4f, theirs %.4f)", ours.landLng, theirs.landLng) + breaks++ + } + case !aboutLongitude(oursErr) || !aboutLongitude(theirsErr): + // A rejection for some other reason says nothing about notation. Before + // this check, an upstream "No matching dataset found" — which happens as + // soon as sondehub rotates past our GFS run — was reported as "Tawhiri + // refuses this longitude", i.e. the tool inventing compatibility news. + verdict = "inconclusive: " + firstReason(oursErr, theirsErr) + case oursErr != nil && theirsErr != nil: + verdict = "agree (both refuse)" + case oursErr != nil: + verdict = "BREAK: Tawhiri accepts, we refuse" + breaks++ + default: + verdict = "extension: we accept, Tawhiri refuses" + } + fmt.Fprintf(tw, "%g\t%s\t%s\t%s\t%s\n", + lng, code(oursStatus, oursErr), code(theirsStatus, theirsErr), delta, verdict) + } + tw.Flush() + return breaks +} + +// aboutLongitude reports whether err is a rejection of the longitude itself +// rather than of something else in the request. A nil error passes: a successful +// call is always informative. +func aboutLongitude(err error) bool { + if err == nil { + return true + } + return strings.Contains(strings.ToLower(err.Error()), "longitude") +} + +// firstReason summarises whichever side failed for an unrelated reason. +func firstReason(errs ...error) string { + for _, err := range errs { + if err != nil && !aboutLongitude(err) { + return truncate(err.Error(), 90) + } + } + return "unknown" +} + +// longitudePairs are two notations naming one meridian: signed, then unsigned. +// +// Tawhiri reads only the unsigned one, so this is the only way the signed form — +// the extension the product actually depends on — can be checked against the +// reference at all. Our answer for the signed value must match Tawhiri's answer +// for its unsigned twin. Without this the "extension" rows in the table above are +// merely unrefuted, not verified. +// +// (0, 360) is absent on purpose: Tawhiri refuses 360, so that pair has no +// reference side. +func longitudePairs() [][2]float64 { + return [][2]float64{{-51.7, 308.3}, {-180, 180}, {-90, 270}, {-0.0001, 359.9999}} +} + +// comparePairs checks each signed notation we accept against Tawhiri's answer for +// the same meridian written the way Tawhiri accepts it. Returns the number of +// pairs that disagree by more than the site suite's own tolerance. +func comparePairs(server, tawhiri string, base params, datasetParam string) int { + tw := tabwriter.NewWriter(os.Stdout, 0, 0, 2, ' ', 0) + fmt.Fprintln(tw, "\nours (signed)\tvs tawhiri (unsigned)\tlanding Δ\tverdict") + fmt.Fprintln(tw, "-------------\t---------------------\t---------\t-------") + + bad := 0 + for _, pair := range longitudePairs() { + signed, unsigned := pair[0], pair[1] + + p := base + p.lng = signed + ours, _, oursErr := predict(server+"/api/v1/prediction", p, "") + + q := base + q.lng = unsigned + theirs, _, theirsErr := predict(tawhiri, q, datasetParam) + + switch { + case oursErr != nil: + fmt.Fprintf(tw, "%g\t%g\t\tours refused: %v\n", signed, unsigned, oursErr) + bad++ + case theirsErr != nil: + fmt.Fprintf(tw, "%g\t%g\t\ttawhiri refused the unsigned twin: %v\n", signed, unsigned, theirsErr) + bad++ + default: + d := haversine(ours.landLat, ours.landLng, theirs.landLat, theirs.landLng) + verdict := "agree — the extension reads the same meridian" + if d > 50000 { + verdict = "DIVERGENT — signed notation is not the same meridian" + bad++ + } + fmt.Fprintf(tw, "%g\t%g\t%.2f km\t%s\n", signed, unsigned, d/1000, verdict) + } + } + tw.Flush() + return bad +} diff --git a/internal/api/mapping.go b/internal/api/mapping.go index e3d4d72..98661f4 100644 --- a/internal/api/mapping.go +++ b/internal/api/mapping.go @@ -2,22 +2,72 @@ package api import ( "fmt" + "math" + "net/http" "time" "predictor-refactored/internal/api/async" "predictor-refactored/internal/engine" + "predictor-refactored/internal/numerics" apirest "predictor-refactored/pkg/rest" ) -// normalizeLng folds a longitude into [0, 360) for internal use. -func normalizeLng(lng float64) float64 { - if lng < 0 { - return lng + 360 +// longitudeLimit bounds what either API version accepts, in degrees. +// +// Symmetric on purpose. Upstream Tawhiri implements a half-open [0, 360) — one +// fundamental domain, every meridian spelled exactly once (measured). We accept a +// superset because the frontend holds longitudes in [-180, 180], so a western +// launch arrives negative, and we tolerate one full turn either way. Symmetry is +// the whole point: the previous bound, [-180, 360), had its two ends justified by +// unrelated things — the lower end was the signed convention's edge, the upper end +// was the domain on which the old single-fold normalizeLng happened to be correct. +// +// Beyond one turn a value is a typo rather than a notation. 5170 is arithmetically +// a perfectly good way to write 50 E, and upstream refuses it too. +const longitudeLimit = 360.0 + +// validateLat refuses latitudes outside [-90, 90]. +// +// Shared by both API versions. Unlike longitude this is a real geometric bound: +// the poles are the ends of the axis, so no notation makes 100 meaningful. v2 +// checked it and v1 did not. +func validateLat(lat float64) error { + if math.IsNaN(lat) || lat < -90 || lat > 90 { + return apiError(http.StatusBadRequest, fmt.Sprintf("launch latitude must be in [-90, 90], got %g", lat)) } - return lng + return nil } -// signedLng converts an internal [0, 360) longitude back to [-180, 180). +// validateLng refuses longitudes outside [-longitudeLimit, longitudeLimit]. +// +// Shared by both API versions. v1 used to apply no check at all while v2 enforced +// [-180, 360), so the same launch could be accepted by one and refused by the +// other — a difference in behaviour where only format may differ. +func validateLng(lng float64) error { + if math.IsNaN(lng) || lng < -longitudeLimit || lng > longitudeLimit { + return apiError(http.StatusBadRequest, + fmt.Sprintf("launch longitude must be in [-%g, %g], got %g", longitudeLimit, longitudeLimit, lng)) + } + return nil +} + +// normalizeLng folds a longitude into the [0, 360) the wind grid is indexed on. +// +// A true modulo, via the same helper the integrator applies on every step +// (numerics.PyMod in toGeo), so the request path and the engine agree by +// construction rather than by coincidence. It used to be a single +// `if lng < 0 { lng += 360 }`, correct only on [-360, 360): -400 became -40, +// reached the grid, and came back as "lng=-40 out of range" — a complaint about +// wind data for what was an input problem. +func normalizeLng(lng float64) float64 { + return numerics.PyMod(lng, 360) +} + +// signedLng converts an internal [0, 360) longitude to [-180, 180). +// +// This is v2's published format. v1 publishes the internal [0, 360) unchanged, +// because that is the convention upstream Tawhiri publishes and v1 exists to be a +// drop-in for it. func signedLng(lng float64) float64 { if lng > 180 { return lng - 360 diff --git a/internal/api/mapping_test.go b/internal/api/mapping_test.go new file mode 100644 index 0000000..37d5265 --- /dev/null +++ b/internal/api/mapping_test.go @@ -0,0 +1,88 @@ +package api + +import ( + "math" + "testing" +) + +// Longitude handling is shared by both API versions on purpose: the two may +// differ in format, names and features, never in what the maths does. v1 used to +// apply no range check at all while v2 enforced [-180, 360), so the same launch +// could be accepted by one and refused by the other. + +func TestValidateLngAcceptsBothConventionsAndOneTurnEitherWay(t *testing.T) { + // Upstream Tawhiri implements a half-open [0, 360) — measured: 0, 180, 270 and + // 308.3 accepted, everything negative and 360 refused. We accept a superset + // because the frontend holds longitudes in [-180, 180], so a western launch + // arrives negative. The bound is symmetric so there is nothing to explain + // about one end that does not hold for the other. + for _, lng := range []float64{0, 0.1, 180, 270, 308.3, 359.999, 360, -0.0001, -51.7, -180, -200, -360} { + if err := validateLng(lng); err != nil { + t.Errorf("validateLng(%g) = %v, want accepted", lng, err) + } + } +} + +func TestValidateLngRefusesTypos(t *testing.T) { + // Arithmetically every one of these names a real meridian — 5170 is 50 E. They + // are refused because they are far likelier to be a unit slip or a swapped + // field than an intended notation, and upstream refuses them too. + for _, lng := range []float64{360.0001, -360.0001, 400, -400, 5170, math.Inf(1), math.Inf(-1)} { + if err := validateLng(lng); err == nil { + t.Errorf("validateLng(%g) was accepted, want refused", lng) + } + } +} + +func TestNormalizeLngFoldsAnyInputIntoTheGridDomain(t *testing.T) { + // The wind grid is indexed from LonStart = 0 over 360 degrees, so every value + // handed to it must land in [0, 360). + // + // This was a single `if lng < 0 { lng += 360 }`, correct only on [-360, 360): + // -400 became -40, reached the grid, and came back as "lng=-40 out of range" — + // a complaint about wind data for what was an input problem. + cases := map[float64]float64{ + 0: 0, 0.5: 0.5, 180: 180, 359.5: 359.5, + 360: 0, 720: 0, -360: 0, + -0.5: 359.5, -51.7: 308.3, -180: 180, -200: 160, -400: 320, 5170: 130, + } + for in, want := range cases { + if got := normalizeLng(in); math.Abs(got-want) > 1e-9 { + t.Errorf("normalizeLng(%g) = %g, want %g", in, got, want) + } + if got := normalizeLng(in); got < 0 || got >= 360 { + t.Errorf("normalizeLng(%g) = %g, outside [0, 360)", in, got) + } + } +} + +func TestSignedLngIsTheV2Format(t *testing.T) { + // v2 publishes [-180, 180): its own convenient format. v1 publishes the + // internal [0, 360) unchanged, because that is what upstream Tawhiri publishes + // and v1 exists to be a drop-in for it. + cases := map[float64]float64{0: 0, 90: 90, 180: 180, 180.5: -179.5, 308.3: -51.7, 359.5: -0.5} + for in, want := range cases { + if got := signedLng(in); math.Abs(got-want) > 1e-9 { + t.Errorf("signedLng(%g) = %g, want %g", in, got, want) + } + } +} + +func TestValidateLatIsSharedByBothVersions(t *testing.T) { + // Unlike longitude, latitude is a genuine geometric bound: +-90 are the poles + // and there is nothing beyond them, so no notation makes 100 meaningful. + // + // v2 checked this and v1 did not, which is a difference in behaviour rather + // than format. On v1 an out-of-range latitude reached the wind grid and came + // back as "lat=... out of range" — a complaint about data for an input problem. + for _, lat := range []float64{-90, -89.999, 0, 52.2, 89.999, 90} { + if err := validateLat(lat); err != nil { + t.Errorf("validateLat(%g) = %v, want accepted", lat, err) + } + } + for _, lat := range []float64{-90.0001, 90.0001, 100, -100, math.NaN(), math.Inf(1)} { + if err := validateLat(lat); err == nil { + t.Errorf("validateLat(%g) was accepted, want refused", lat) + } + } +} diff --git a/internal/api/prediction.go b/internal/api/prediction.go index 5af501d..2399270 100644 --- a/internal/api/prediction.go +++ b/internal/api/prediction.go @@ -2,6 +2,7 @@ package api import ( "context" + "fmt" "net/http" "time" @@ -23,9 +24,45 @@ func (h *Handler) ReadinessCheck(_ context.Context) (*apirest.ReadinessResponse, return resp, nil } +// fieldFor resolves the dataset a request asked for, or the active one. +// +// A request naming a dataset that is not stored is refused. The parameter used +// to be declared in the spec and never read, so a client asking for a specific +// forecast run — an archival date, a reproducible comparison — was silently +// served from whatever happened to be loaded instead. +func (h *Handler) fieldFor(ctx context.Context, want apirest.OptDateTime) (weather.WindField, error) { + if epoch, ok := want.Get(); ok { + field, err := h.mgr.FieldFor(ctx, epoch.UTC()) + if err != nil { + return nil, apiError(http.StatusBadRequest, err.Error()) + } + return field, nil + } + field := h.mgr.Active() + if field == nil { + return nil, apiError(http.StatusServiceUnavailable, "no dataset loaded, service is starting up") + } + return field, nil +} + +// runFailure turns an integration failure into a 400. +// +// The sampler reporting an error means it had no data — most often a launch time +// past the dataset's horizon — so the trajectory is not a forecast. It used to be +// discarded, and the run continued with zero wind: the balloon took off and +// landed on the same spot, returned as a successful prediction. +func runFailure(events *engine.EventSink, field weather.WindField) error { + err := events.Err() + if err == nil { + return nil + } + return apiError(http.StatusBadRequest, + fmt.Sprintf("%s (dataset %s)", err, field.Epoch().UTC().Format(time.RFC3339))) +} + // PerformPredictionV2 implements POST /api/v2/prediction. -func (h *Handler) PerformPredictionV2(_ context.Context, req *apirest.PredictionV2Request) (*apirest.PredictionV2Response, error) { - resp, err := h.runPredictionV2(req) +func (h *Handler) PerformPredictionV2(ctx context.Context, req *apirest.PredictionV2Request) (*apirest.PredictionV2Response, error) { + resp, err := h.runPredictionV2(ctx, req) if err == nil { h.metrics.Prediction("v2", resp.CompletedAt.Sub(resp.StartedAt), nil) } @@ -60,23 +97,22 @@ func (h *Handler) CancelPredictionJob(_ context.Context, params apirest.CancelPr // runPredictionV2 is the synchronous prediction core, shared by the v2 // endpoint and the async worker pool. -func (h *Handler) runPredictionV2(req *apirest.PredictionV2Request) (*apirest.PredictionV2Response, error) { +func (h *Handler) runPredictionV2(ctx context.Context, req *apirest.PredictionV2Request) (*apirest.PredictionV2Response, error) { // Validate the request shape before checking dataset availability, so a // malformed request is a 400 regardless of startup state. lat := req.Launch.Latitude - rawLng := req.Launch.Longitude alt := req.Launch.Altitude.Or(0) - if lat < -90 || lat > 90 { - return nil, apiError(http.StatusBadRequest, "launch.latitude must be in [-90, 90]") + if err := validateLat(lat); err != nil { + return nil, err } - if rawLng < -180 || rawLng >= 360 { - return nil, apiError(http.StatusBadRequest, "launch.longitude must be in [-180, 360)") + if err := validateLng(req.Launch.Longitude); err != nil { + return nil, err } - lng := normalizeLng(rawLng) + lng := normalizeLng(req.Launch.Longitude) - field := h.mgr.Active() - if field == nil { - return nil, apiError(http.StatusServiceUnavailable, "no dataset loaded, service is starting up") + field, err := h.fieldFor(ctx, req.Dataset) + if err != nil { + return nil, err } events := engine.NewEventSink() @@ -90,6 +126,9 @@ func (h *Handler) runPredictionV2(req *apirest.PredictionV2Request) (*apirest.Pr started := time.Now().UTC() results := prof.Run(float64(req.Launch.Time.Unix()), engine.State{Lat: lat, Lng: lng, Altitude: alt}, events) completed := time.Now().UTC() + if err := runFailure(events, field); err != nil { + return nil, err + } resp := &apirest.PredictionV2Response{ Stages: make([]apirest.StageResult, 0, len(results)), @@ -105,10 +144,10 @@ func (h *Handler) runPredictionV2(req *apirest.PredictionV2Request) (*apirest.Pr } // PerformPrediction implements GET /api/v1/prediction (Tawhiri-compatible). -func (h *Handler) PerformPrediction(_ context.Context, params apirest.PerformPredictionParams) (*apirest.PredictionResponse, error) { - field := h.mgr.Active() - if field == nil { - return nil, apiError(http.StatusServiceUnavailable, "no dataset loaded, service is starting up") +func (h *Handler) PerformPrediction(ctx context.Context, params apirest.PerformPredictionParams) (*apirest.PredictionResponse, error) { + field, err := h.fieldFor(ctx, params.Dataset) + if err != nil { + return nil, err } profileKind := "standard_profile" @@ -118,6 +157,12 @@ func (h *Handler) PerformPrediction(_ context.Context, params apirest.PerformPre ascentRate := params.AscentRate.Or(5) descentRate := params.DescentRate.Or(5) launchAlt := params.LaunchAltitude.Or(0) + if err := validateLat(params.LaunchLatitude); err != nil { + return nil, err + } + if err := validateLng(params.LaunchLongitude); err != nil { + return nil, err + } lng := normalizeLng(params.LaunchLongitude) launchTime := float64(params.LaunchDatetime.Unix()) @@ -142,6 +187,9 @@ func (h *Handler) PerformPrediction(_ context.Context, params apirest.PerformPre started := time.Now().UTC() results := prof.Run(launchTime, engine.State{Lat: params.LaunchLatitude, Lng: lng, Altitude: launchAlt}, events) completed := time.Now().UTC() + if err := runFailure(events, field); err != nil { + return nil, err + } h.metrics.Prediction(profileKind, completed.Sub(started), nil) resp := &apirest.PredictionResponse{ @@ -229,9 +277,13 @@ func tawhiriItem(name string, r engine.Result) apirest.PredictionResponsePredict for i := range n { t, p := r.Path.At(i) traj = append(traj, apirest.TawhiriPoint{ - Datetime: time.Unix(int64(t), 0).UTC(), - Latitude: p.Lat, - Longitude: signedLng(p.Lng), + Datetime: time.Unix(int64(t), 0).UTC(), + Latitude: p.Lat, + // v1 publishes [0, 360), the convention upstream Tawhiri publishes — + // measured: it answers 308.7884 where we used to answer -51.2115 for + // the same meridian. v1 exists to be a drop-in, so it matches. v2 keeps + // the signed format (mapping.go). + Longitude: p.Lng, Altitude: p.Altitude, }) } diff --git a/internal/api/transport.go b/internal/api/transport.go index 75a9122..fe23987 100644 --- a/internal/api/transport.go +++ b/internal/api/transport.go @@ -74,7 +74,11 @@ func New(port int, d Deps) (*Server, error) { Workers: d.AsyncWorkers, QueueSize: d.AsyncQueueSize, ResultTTL: d.AsyncResultTTL, - }, h.runPredictionV2, d.Metrics, d.Log) + // The async queue outlives the HTTP request that enqueued the job, so a + // worker cannot inherit its context. + }, func(req *apirest.PredictionV2Request) (*apirest.PredictionV2Response, error) { + return h.runPredictionV2(context.Background(), req) + }, d.Metrics, d.Log) ogenSrv, err := apirest.NewServer(h, apirest.WithMiddleware(middleware.OgenLogging(d.Log))) if err != nil { diff --git a/internal/datasets/manager.go b/internal/datasets/manager.go index b56a265..2881545 100644 --- a/internal/datasets/manager.go +++ b/internal/datasets/manager.go @@ -161,6 +161,57 @@ func (m *Manager) SelectFor(t time.Time, lat, lng float64) weather.WindField { return nil } +// FieldFor returns the global dataset whose epoch is epoch, loading it from +// storage on demand when it is stored but not currently active. +// +// Being stored is enough. A caller asking for a specific run — an archival +// forecast, a reproducible comparison — must not have to make it the service's +// active dataset first, and asking for it must not repoint the service: Load +// appends, and Active keeps returning the first global it finds. +// +// Errors when no stored dataset carries that epoch, so a request naming a +// dataset the service cannot serve is refused rather than quietly answered from +// a different one. +// +// ponytail: loaded datasets are never evicted, so a long-lived service asked for +// many distinct epochs accumulates one mmap and fd each. Add LRU eviction here +// if that becomes real; the files are mmap-backed, so the cost is address space +// and descriptors, not resident memory. +func (m *Manager) FieldFor(ctx context.Context, epoch time.Time) (weather.WindField, error) { + if f := m.activeFieldAt(epoch); f != nil { + return f, nil + } + + stored, err := m.store.List() + if err != nil { + return nil, fmt.Errorf("list stored datasets: %w", err) + } + for _, id := range stored { + if !id.Subset.IsGlobal() || !id.Epoch.Equal(epoch) { + continue + } + if err := m.Load(ctx, id); err != nil { + return nil, fmt.Errorf("load %s: %w", id.Filename(), err) + } + if f := m.activeFieldAt(epoch); f != nil { + return f, nil + } + } + return nil, fmt.Errorf("dataset %s is not stored", epoch.UTC().Format(time.RFC3339)) +} + +// activeFieldAt returns the loaded global field with exactly this epoch, or nil. +func (m *Manager) activeFieldAt(epoch time.Time) weather.WindField { + m.activeMu.RLock() + defer m.activeMu.RUnlock() + for _, d := range m.active { + if d.ID.Subset.IsGlobal() && d.ID.Epoch.Equal(epoch) { + return d.Field + } + } + return nil +} + // LoadedDatasets returns snapshots of every currently-loaded dataset. func (m *Manager) LoadedDatasets() []LoadedDatasetInfo { m.activeMu.RLock() @@ -309,36 +360,48 @@ func (m *Manager) Load(ctx context.Context, id DatasetID) error { // // Returns the JobID started, or empty string when nothing was scheduled. func (m *Manager) Refresh(ctx context.Context, freshnessTTL time.Duration) (string, error) { - if a := m.activeGlobal(); a != nil && time.Since(a.ID.Epoch) < freshnessTTL { - return "", nil - } - - if datasets, err := m.store.List(); err == nil { - for _, id := range datasets { - if !id.Subset.IsGlobal() { - continue - } - if time.Since(id.Epoch) > freshnessTTL { - continue - } - if a := m.activeGlobal(); a != nil && a.ID.Equals(id) { - return "", nil - } - if err := m.Load(ctx, id); err == nil { - return "", nil + // Get something usable loaded first, whatever its age. + // + // freshnessTTL answers one question only: go fetch something newer? It must + // never decide whether data already on disk may be read. While one check + // served both, a stored run older than the TTL was skipped even when it was + // the only dataset present — active stayed empty, every prediction answered + // "no dataset loaded" with gigabytes of usable wind on disk, and with no + // reachable origin that state was permanent. + if m.activeGlobal() == nil { + if stored, err := m.store.List(); err == nil { + for _, id := range stored { // newest first + if !id.Subset.IsGlobal() { + continue + } + if err := m.Load(ctx, id); err == nil { + break + } } } } + active := m.activeGlobal() + if active != nil && time.Since(active.ID.Epoch) < freshnessTTL { + return "", nil + } + latest, err := m.src.LatestEpoch(ctx) if err != nil { return "", fmt.Errorf("latest epoch: %w", err) } - id := DatasetID{Epoch: latest} - if a := m.activeGlobal(); a != nil && !latest.After(a.ID.Epoch) { + if active != nil && !latest.After(active.ID.Epoch) { return "", nil } + // Another replica sharing this volume may have committed it already. + id := DatasetID{Epoch: latest} + if m.store.Exists(id) { + if err := m.Load(ctx, id); err == nil { + return "", nil + } + } + jobID := m.Download(id) go m.loadAfterCompletion(jobID, id) return jobID, nil diff --git a/internal/datasets/manager_test.go b/internal/datasets/manager_test.go new file mode 100644 index 0000000..c68f724 --- /dev/null +++ b/internal/datasets/manager_test.go @@ -0,0 +1,166 @@ +package datasets + +import ( + "context" + "errors" + "testing" + "time" + + "predictor-refactored/internal/weather" +) + +// stubField is a WindField that carries only its epoch. +type stubField struct{ epoch time.Time } + +func (f stubField) Wind(_, _, _, _ float64) (weather.Sample, error) { + return weather.Sample{}, nil +} +func (f stubField) Epoch() time.Time { return f.epoch } +func (f stubField) Source() string { return "fake" } + +// fakeSource records what it was asked to open and can simulate an unreachable +// origin via latestErr. +type fakeSource struct { + latest time.Time + latestErr error + opened []DatasetID +} + +func (s *fakeSource) ID() string { return "fake" } + +func (s *fakeSource) LatestEpoch(context.Context) (time.Time, error) { + if s.latestErr != nil { + return time.Time{}, s.latestErr + } + return s.latest, nil +} + +func (s *fakeSource) Download(context.Context, DatasetID, Storage, ProgressSink, Throttle) error { + return errors.New("fakeSource does not download") +} + +func (s *fakeSource) Open(_ context.Context, id DatasetID, _ Storage) (weather.WindField, error) { + s.opened = append(s.opened, id) + return stubField{epoch: id.Epoch}, nil +} + +func (s *fakeSource) Coverage(id DatasetID) Coverage { + return Coverage{ + Region: Region{MinLat: -90, MaxLat: 90, MinLng: 0, MaxLng: 360}, + StartTime: id.Epoch, + EndTime: id.Epoch.Add(192 * time.Hour), + } +} + +// fakeStore is an in-memory Storage holding a fixed set of committed datasets, +// newest first, as LocalStore.List promises. +type fakeStore struct{ ids []DatasetID } + +func (s *fakeStore) SourceID() string { return "fake" } +func (s *fakeStore) Path(id DatasetID) string { return "/dev/null/" + id.Filename() } + +func (s *fakeStore) Exists(id DatasetID) bool { + for _, have := range s.ids { + if have.Equals(id) { + return true + } + } + return false +} + +func (s *fakeStore) List() ([]DatasetID, error) { return s.ids, nil } +func (s *fakeStore) Remove(DatasetID) error { return nil } +func (s *fakeStore) BeginWrite(DatasetID) (TempHandle, error) { return nil, errors.New("not used") } +func (s *fakeStore) Lock(context.Context) (func(), error) { return func() {}, nil } + +// TestRefreshLoadsAStoredDatasetOlderThanTheFreshnessTTL pins the distinction +// the freshness TTL is allowed to make. +// +// A dataset on disk is usable wind data whatever its age. The TTL answers "go +// fetch something newer?", not "may I read what is already here?". While both +// questions shared one check, an archival run that was the only dataset present +// got skipped, active stayed empty, and every prediction answered "no dataset +// loaded" with gigabytes of usable data on disk — and, with no reachable origin, +// forever. +func TestRefreshLoadsAStoredDatasetOlderThanTheFreshnessTTL(t *testing.T) { + old := time.Now().UTC().Add(-30 * 24 * time.Hour).Truncate(time.Hour) + src := &fakeSource{latestErr: errors.New("origin unreachable")} + store := &fakeStore{ids: []DatasetID{{Epoch: old}}} + m := New(src, store, nil, nil) + + // The origin probe is expected to fail; loading what is on disk is not. + _, _ = m.Refresh(context.Background(), 48*time.Hour) + + if m.Active() == nil { + t.Fatal("Active() is nil: the stored dataset was never loaded") + } + if got := m.Active().Epoch(); !got.Equal(old) { + t.Errorf("loaded epoch = %s, want %s", got, old) + } +} + +// TestRefreshPrefersTheNewestStoredDataset guards the ordering the fix relies +// on: falling back to disk must not mean falling back to the oldest file there. +func TestRefreshPrefersTheNewestStoredDataset(t *testing.T) { + newer := time.Now().UTC().Add(-10 * 24 * time.Hour).Truncate(time.Hour) + older := newer.Add(-20 * 24 * time.Hour) + src := &fakeSource{latestErr: errors.New("origin unreachable")} + store := &fakeStore{ids: []DatasetID{{Epoch: newer}, {Epoch: older}}} + m := New(src, store, nil, nil) + + _, _ = m.Refresh(context.Background(), 48*time.Hour) + + if m.Active() == nil { + t.Fatal("Active() is nil") + } + if got := m.Active().Epoch(); !got.Equal(newer) { + t.Errorf("loaded epoch = %s, want the newest stored %s", got, newer) + } +} + +// TestFieldForLoadsAStoredDatasetThatIsNotActive covers asking for a specific +// run by epoch — the archival-forecast case. Being stored is enough; the caller +// should not have to make it the active dataset first. +func TestFieldForLoadsAStoredDatasetThatIsNotActive(t *testing.T) { + active := time.Now().UTC().Truncate(time.Hour) + archive := active.Add(-30 * 24 * time.Hour) + src := &fakeSource{latest: active} + store := &fakeStore{ids: []DatasetID{{Epoch: active}, {Epoch: archive}}} + m := New(src, store, nil, nil) + + if err := m.Load(context.Background(), DatasetID{Epoch: active}); err != nil { + t.Fatalf("Load active: %v", err) + } + + field, err := m.FieldFor(context.Background(), archive) + if err != nil { + t.Fatalf("FieldFor(%s): %v", archive, err) + } + if got := field.Epoch(); !got.Equal(archive) { + t.Errorf("epoch = %s, want %s", got, archive) + } + // The active dataset must be unaffected: asking for an archival run for one + // prediction should not repoint the service at it. + if got := m.Active().Epoch(); !got.Equal(active) { + t.Errorf("Active() moved to %s, want %s", got, active) + } +} + +// TestFieldForRejectsAnEpochThatIsNotStored is the half that matters most: a +// requested dataset that cannot be served must say so rather than quietly +// substituting a different one, which is what ignoring the parameter did. +func TestFieldForRejectsAnEpochThatIsNotStored(t *testing.T) { + active := time.Now().UTC().Truncate(time.Hour) + src := &fakeSource{latest: active} + store := &fakeStore{ids: []DatasetID{{Epoch: active}}} + m := New(src, store, nil, nil) + if err := m.Load(context.Background(), DatasetID{Epoch: active}); err != nil { + t.Fatalf("Load: %v", err) + } + + missing := active.Add(-365 * 24 * time.Hour) + field, err := m.FieldFor(context.Background(), missing) + if err == nil { + t.Fatalf("FieldFor(%s) returned field with epoch %s, want an error", missing, field.Epoch()) + } +} diff --git a/internal/engine/engine_test.go b/internal/engine/engine_test.go index ccf0f88..9344f7d 100644 --- a/internal/engine/engine_test.go +++ b/internal/engine/engine_test.go @@ -1,7 +1,9 @@ package engine import ( + "fmt" "math" + "strings" "testing" "time" @@ -261,3 +263,157 @@ func TestPolygonOutsideAntimeridian(t *testing.T) { t.Errorf("(0, 0) should be outside") } } + +// horizonWind has data up to horizon seconds and fails past it, the way a real +// dataset's time axis does (numerics.AxisError from Axis.Locate). +type horizonWind struct{ horizon float64 } + +func (w horizonWind) Wind(t, _, _, _ float64) (weather.Sample, error) { + if t > w.horizon { + return weather.Sample{}, fmt.Errorf("hour=%v out of range", t/3600) + } + return weather.Sample{U: 10, V: 0}, nil +} +func (w horizonWind) Epoch() time.Time { return time.Unix(0, 0).UTC() } +func (w horizonWind) Source() string { return "test" } + +// TestWindSamplingFailureIsRecordedNotSwallowed pins the difference between an +// event and a failure. +// +// WindTransport used to discard the sampler's error and return a zero rate. A +// zero rate is indistinguishable from calm air, so a launch past the dataset's +// horizon integrated with no wind at all and produced a balloon that took off +// and landed on the same spot — returned as a successful forecast. An error from +// the sampler means there was no data, so the trajectory past that point is not +// a forecast and must not be presented as one. +func TestWindSamplingFailureIsRecordedNotSwallowed(t *testing.T) { + sink := NewEventSink() + ascend := &Propagator{ + Name: "ascent", + Step: 60, + Model: Sum(ConstantRate(5), WindTransport(horizonWind{horizon: 600}, sink)), + Constraints: []Constraint{Altitude{Op: OpGreaterEqual, Limit: 30000, On: ActionStop}}, + } + prof := Profile{Stages: []*Propagator{ascend}, Direction: Forward} + prof.Run(0, State{Lat: 0, Lng: 0, Altitude: 0}, sink) + + err := sink.Err() + if err == nil { + t.Fatal("sink reports no failure; the sampling error was swallowed") + } + if !strings.Contains(err.Error(), "out of range") { + t.Errorf("error %q does not carry the sampler's reason", err) + } +} + +// TestWindSamplingSuccessLeavesNoFailure keeps the check above from firing on +// healthy runs, which would turn every prediction into a 400. +func TestWindSamplingSuccessLeavesNoFailure(t *testing.T) { + sink := NewEventSink() + ascend := &Propagator{ + Name: "ascent", + Step: 60, + Model: Sum(ConstantRate(5), WindTransport(horizonWind{horizon: 1e9}, sink)), + Constraints: []Constraint{Altitude{Op: OpGreaterEqual, Limit: 30000, On: ActionStop}}, + } + prof := Profile{Stages: []*Propagator{ascend}, Direction: Forward} + prof.Run(0, State{Lat: 0, Lng: 0, Altitude: 0}, sink) + + if err := sink.Err(); err != nil { + t.Fatalf("healthy run reported a failure: %v", err) + } +} + +// TestAboveModelStaysNonFatal separates the two paths explicitly: extrapolating +// above the highest pressure level is a warning the caller may ignore, and must +// not become a hard failure. +func TestAboveModelStaysNonFatal(t *testing.T) { + sink := NewEventSink() + ascend := &Propagator{ + Name: "ascent", + Step: 60, + Model: Sum(ConstantRate(5), WindTransport(aboveModelWind{}, sink)), + Constraints: []Constraint{Altitude{Op: OpGreaterEqual, Limit: 30000, On: ActionStop}}, + } + prof := Profile{Stages: []*Propagator{ascend}, Direction: Forward} + prof.Run(0, State{Lat: 0, Lng: 0, Altitude: 0}, sink) + + if err := sink.Err(); err != nil { + t.Fatalf("above_model became a failure: %v", err) + } + if len(sink.Snapshot()) == 0 { + t.Error("above_model event was not emitted") + } +} + +// TestEveryModelHonoursIncludeWind pins the promise ModelSpec.IncludeWind makes. +// +// buildConstantRate and buildParachuteDescent both took BuildDeps as `_` and +// returned a bare vertical model, so include_wind was accepted and discarded. +// Only buildPiecewise ever called maybeAddWind. POST /api/v2/prediction with the +// spec's own documented example therefore answered 200 with a trajectory that +// rose and fell on the spot — a wind-free forecast presented as a forecast. +// +// The loop over modelFactories is the part that matters going forward: a newly +// registered model type cannot be added without deciding here what include_wind +// does to it, which is exactly the step that was skipped before. +func TestEveryModelHonoursIncludeWind(t *testing.T) { + cases := map[string]struct { + spec ModelSpec + wantError bool + }{ + "constant_rate": {spec: ModelSpec{Type: "constant_rate", Rate: 5, IncludeWind: true}}, + "parachute_descent": {spec: ModelSpec{Type: "parachute_descent", SeaLevelRate: 5, IncludeWind: true}}, + "piecewise": {spec: ModelSpec{Type: "piecewise", IncludeWind: true, + Segments: []PiecewiseSegmentSpec{{Until: math.Inf(1), Rate: 5}}}}, + // include_wind on the wind model itself would sum the wind into itself. + // Refused rather than ignored: a silently doubled wind is a wrong forecast + // that looks right. + "wind": {spec: ModelSpec{Type: "wind", IncludeWind: true}, wantError: true}, + } + + for name := range modelFactories { + if _, ok := cases[name]; !ok { + t.Errorf("model %q is registered but has no include_wind case here", name) + } + } + + deps := BuildDeps{Wind: fixedWind{u: 10, v: -4}} + for name, tc := range cases { + t.Run(name, func(t *testing.T) { + built, err := BuildModel(tc.spec, deps) + if tc.wantError { + if err == nil { + t.Fatal("expected include_wind to be refused, got no error") + } + return + } + if err != nil { + t.Fatalf("BuildModel: %v", err) + } + m := built.Model + if built.Build != nil { + m = built.Build(StageContext{}) + } + rate := m(0, State{Lat: 0, Lng: 0, Altitude: 1000}) + if rate.East != 10 || rate.North != -4 { + t.Errorf("include_wind ignored: east/north = %v/%v, want 10/-4", rate.East, rate.North) + } + }) + } +} + +// TestIncludeWindWithoutAFieldIsRefused covers the other half of maybeAddWind, +// which returned the wind-free model when deps.Wind was nil. A request that asks +// for wind and cannot get it must fail, not quietly become a vertical drop. +func TestIncludeWindWithoutAFieldIsRefused(t *testing.T) { + for _, typ := range []string{"constant_rate", "parachute_descent", "piecewise"} { + t.Run(typ, func(t *testing.T) { + spec := ModelSpec{Type: typ, Rate: 5, SeaLevelRate: 5, IncludeWind: true, + Segments: []PiecewiseSegmentSpec{{Until: math.Inf(1), Rate: 5}}} + if _, err := BuildModel(spec, BuildDeps{}); err == nil { + t.Error("include_wind with no wind field was accepted") + } + }) + } +} diff --git a/internal/engine/events.go b/internal/engine/events.go index 7fde684..3032320 100644 --- a/internal/engine/events.go +++ b/internal/engine/events.go @@ -26,10 +26,13 @@ type EventSummary struct { } // EventSink collects events from models and the integrator, aggregating -// duplicate types into a single EventSummary. Safe for concurrent use. +// duplicate types into a single EventSummary. It also carries the run's first +// unrecoverable failure, which is a different thing from an event — see Fail. +// Safe for concurrent use. type EventSink struct { mu sync.Mutex summaries map[string]*EventSummary + err error } // NewEventSink returns an empty sink. @@ -61,6 +64,42 @@ func (s *EventSink) Emit(typ string, t float64, state State, message string) { } } +// Fail records the run's first unrecoverable error. Nil-safe, like Emit. +// +// Deliberately not an event. An event is an observation the caller may choose to +// ignore: "above_model" means samples above the highest pressure level were +// extrapolated, which degrades the answer without invalidating it. A failure +// means the sampler had no data at all, so the trajectory from that point on is +// not a forecast and must not be returned as one. +// +// Routing both through Emit is how the second came to be treated like the first. +// WindTransport discarded the sampler's error and returned a zero rate, which is +// indistinguishable from calm air, so any launch past the dataset's horizon +// integrated with no wind and produced a balloon that took off and landed on the +// same spot — reported as a successful prediction. +// +// Only the first failure is kept: it is the one that explains the run. +func (s *EventSink) Fail(err error) { + if s == nil || err == nil { + return + } + s.mu.Lock() + defer s.mu.Unlock() + if s.err == nil { + s.err = err + } +} + +// Err returns the first failure passed to Fail, or nil if the run was sound. +func (s *EventSink) Err() error { + if s == nil { + return nil + } + s.mu.Lock() + defer s.mu.Unlock() + return s.err +} + // Snapshot returns a stable copy of every summary in deterministic order // (sorted by Type). func (s *EventSink) Snapshot() []EventSummary { diff --git a/internal/engine/models.go b/internal/engine/models.go index 7399961..bb9e72b 100644 --- a/internal/engine/models.go +++ b/internal/engine/models.go @@ -1,7 +1,9 @@ package engine import ( + "fmt" "sort" + "time" "predictor-refactored/internal/numerics" "predictor-refactored/internal/weather" @@ -84,6 +86,11 @@ func WindTransport(field weather.WindField, events *EventSink) Model { return func(t float64, s State) numerics.Rate { sample, err := field.Wind(t, s.Lat, s.Lng, s.Altitude) if err != nil { + // Recorded, not swallowed. A zero rate here is indistinguishable + // from calm air, so returning one turns "we have no data for this + // time" into "the balloon did not move" — see EventSink.Fail. + events.Fail(fmt.Errorf("no wind data at %s: %w", + time.Unix(int64(t), 0).UTC().Format(time.RFC3339), err)) return numerics.Rate{} } if sample.AboveModel && events != nil { diff --git a/internal/engine/registry.go b/internal/engine/registry.go index 27a7a9f..9676ddc 100644 --- a/internal/engine/registry.go +++ b/internal/engine/registry.go @@ -205,21 +205,32 @@ func buildPolygon(spec ConstraintSpec, _ BuildDeps) (Constraint, error) { return NewPolygon(spec.Vertices, mode, act, spec.Label), nil } -func buildConstantRate(spec ModelSpec, _ BuildDeps) (BuiltModel, error) { - return BuiltModel{Model: ConstantRate(spec.Rate)}, nil +func buildConstantRate(spec ModelSpec, deps BuildDeps) (BuiltModel, error) { + if err := checkWind(spec.IncludeWind, deps); err != nil { + return BuiltModel{}, err + } + return BuiltModel{Model: addWind(ConstantRate(spec.Rate), spec.IncludeWind, deps)}, nil } -func buildParachuteDescent(spec ModelSpec, _ BuildDeps) (BuiltModel, error) { +func buildParachuteDescent(spec ModelSpec, deps BuildDeps) (BuiltModel, error) { if spec.SeaLevelRate <= 0 { return BuiltModel{}, fmt.Errorf("parachute_descent requires positive sea_level_rate") } - return BuiltModel{Model: ParachuteDescent(spec.SeaLevelRate)}, nil + if err := checkWind(spec.IncludeWind, deps); err != nil { + return BuiltModel{}, err + } + return BuiltModel{Model: addWind(ParachuteDescent(spec.SeaLevelRate), spec.IncludeWind, deps)}, nil } -func buildWind(_ ModelSpec, deps BuildDeps) (BuiltModel, error) { +func buildWind(spec ModelSpec, deps BuildDeps) (BuiltModel, error) { if deps.Wind == nil { return BuiltModel{}, fmt.Errorf("wind model requires a loaded wind field") } + // Refused rather than ignored: summing the wind into itself would double the + // drift, and a doubled wind is a wrong forecast that looks like a right one. + if spec.IncludeWind { + return BuiltModel{}, fmt.Errorf("wind model does not take include_wind: it would sum the wind into itself") + } return BuiltModel{Model: WindTransport(deps.Wind, deps.Events)}, nil } @@ -231,12 +242,17 @@ func buildPiecewise(spec ModelSpec, deps BuildDeps) (BuiltModel, error) { return BuiltModel{}, fmt.Errorf("piecewise: unknown segment reference %q", s.Reference) } } + // Checked here, not in the closure below: the profile runner calls Build once + // per stage and has nowhere to return an error to. + if err := checkWind(spec.IncludeWind, deps); err != nil { + return BuiltModel{}, err + } // Always build lazily: the profile runner supplies a StageContext before // each stage, which is what resolves absolute / profile-relative / // propagator-relative segment times uniformly. return BuiltModel{ Build: func(ctx StageContext) Model { - return maybeAddWind(Piecewise(resolveSegments(spec.Segments, ctx)), spec.IncludeWind, deps) + return addWind(Piecewise(resolveSegments(spec.Segments, ctx)), spec.IncludeWind, deps) }, }, nil } @@ -266,12 +282,26 @@ func segmentBase(reference string, ctx StageContext) float64 { } } -// maybeAddWind sums a WindTransport model into base when the spec asks for it. -func maybeAddWind(base Model, includeWind bool, deps BuildDeps) Model { - if !includeWind { - return base +// checkWind reports whether include_wind can be satisfied at all. +// +// A request that asks for wind and cannot have it is an error. This used to +// return the wind-free model instead, which turned a missing wind field into a +// balloon that rose and fell on the spot, reported as a successful forecast. +func checkWind(includeWind bool, deps BuildDeps) error { + if includeWind && deps.Wind == nil { + return fmt.Errorf("include_wind requires a loaded wind field") } - if deps.Wind == nil { + return nil +} + +// addWind sums a WindTransport model into base when the spec asks for it. +// +// Callers must have called checkWind first. The split exists because the +// piecewise builder combines inside a closure the profile runner invokes per +// stage, with no way to surface an error — so validation happens at build time +// and combination stays infallible, rather than an error being swallowed there. +func addWind(base Model, includeWind bool, deps BuildDeps) Model { + if !includeWind { return base } return Sum(base, WindTransport(deps.Wind, deps.Events)) diff --git a/internal/windviz/windviz.go b/internal/windviz/windviz.go index bec386a..fcf9117 100644 --- a/internal/windviz/windviz.go +++ b/internal/windviz/windviz.go @@ -68,9 +68,20 @@ const ( // Rasterize samples field over req and returns the U/V grid payload. // // Data is laid out in wind-js scan order: row 0 is the northernmost -// latitude (la1), each row runs west→east, longitudes increasing. Per-cell -// sampling errors (e.g. altitude outside the model) are written as 0 rather -// than failing the whole request; a time outside coverage is a hard error. +// latitude (la1), each row runs west→east, longitudes increasing. +// +// Cells that cannot be sampled individually — a regional dataset queried outside +// its region — are written as 0 and the request still succeeds, because a partly +// covered grid is worth drawing. A grid in which nothing could be sampled is an +// error instead. Time is one value for the whole request, so a time past the +// dataset's horizon fails every cell, and returning that as zeros made a velocity +// layer draw an atmosphere at perfect rest where the honest answer is "we have no +// data for this time". +// +// Known gap: an individual zero-filled cell is indistinguishable from genuine +// calm in the payload. The wind-js format expresses missing data as null, which +// would mean typing Data as []*float64; there is no consumer to validate that +// against since the browser wind layer was removed in the Cesium migration. func Rasterize(field weather.WindField, req Request) (Field, error) { step := req.Step if step <= 0 { @@ -125,6 +136,8 @@ func Rasterize(field weather.WindField, req Request) (Field, error) { v := make([]float64, nx*ny) // Row 0 = north (la1); rows descend in latitude. + var failed int + var firstErr error for j := range ny { lat := maxLat - float64(j)*step for i := range nx { @@ -132,12 +145,20 @@ func Rasterize(field weather.WindField, req Request) (Field, error) { s, err := field.Wind(req.Time, lat, normLng(lng), req.Altitude) idx := j*nx + i if err != nil { - continue // leave as 0 + failed++ + if firstErr == nil { + firstErr = err + } + continue // leave as 0; see the doc comment on this distinction } u[idx] = s.U v[idx] = s.V } } + if failed == nx*ny { + return nil, fmt.Errorf("no wind data anywhere in the requested grid at %s: %w", + time.Unix(int64(req.Time), 0).UTC().Format(time.RFC3339), firstErr) + } refTime := time.Unix(int64(req.Time), 0).UTC().Format("2006-01-02T15:04:05.000Z") mk := func(num int, name string, data []float64) Component { diff --git a/internal/windviz/windviz_test.go b/internal/windviz/windviz_test.go index 521cbdb..aace48e 100644 --- a/internal/windviz/windviz_test.go +++ b/internal/windviz/windviz_test.go @@ -1,6 +1,8 @@ package windviz import ( + "fmt" + "strings" "testing" "time" @@ -94,3 +96,86 @@ func TestCacheRoundTrip(t *testing.T) { t.Errorf("cache should hit after put") } } + +// horizonWind has data up to horizon seconds and fails past it, the way a real +// dataset's time axis does. +type horizonWind struct{ horizon float64 } + +func (w horizonWind) Wind(t, _, _, _ float64) (weather.Sample, error) { + if t > w.horizon { + return weather.Sample{}, fmt.Errorf("hour=%v out of range", t/3600) + } + return weather.Sample{U: 7, V: -3}, nil +} +func (w horizonWind) Epoch() time.Time { return time.Unix(0, 0).UTC() } +func (w horizonWind) Source() string { return "test" } + +// northOnlyWind has data only in the northern hemisphere, the way a regional +// subset does. Its failures are genuinely per-cell. +type northOnlyWind struct{} + +func (northOnlyWind) Wind(_, lat, _, _ float64) (weather.Sample, error) { + if lat < 0 { + return weather.Sample{}, fmt.Errorf("lat=%v outside region", lat) + } + return weather.Sample{U: 5, V: 1}, nil +} +func (northOnlyWind) Epoch() time.Time { return time.Unix(0, 0).UTC() } +func (northOnlyWind) Source() string { return "test" } + +// TestRasterizeRefusesWhenNoCellHasData is the case the package comment already +// promised and the code never implemented ("a time outside coverage is a hard +// error"). +// +// Time is one value for the whole request, so a time past the dataset's horizon +// fails every cell. Each failure was written as a zero, so the response was a +// complete grid of zero wind — a velocity layer draws that as an atmosphere at +// perfect rest. Missing data must not be rendered as calm air. +func TestRasterizeRefusesWhenNoCellHasData(t *testing.T) { + f := horizonWind{horizon: 600} + + out, err := Rasterize(f, Request{Time: 700, MinLng: 0, MaxLng: 360, Step: 90}) + + if err == nil { + t.Fatalf("Rasterize returned %d components instead of an error", len(out)) + } + if !strings.Contains(err.Error(), "out of range") { + t.Errorf("error %q does not carry the sampler's reason", err) + } +} + +// TestRasterizeStillSucceedsWithinCoverage keeps the check above from rejecting +// healthy requests. +func TestRasterizeStillSucceedsWithinCoverage(t *testing.T) { + f := horizonWind{horizon: 600} + + out, err := Rasterize(f, Request{Time: 300, MinLng: 0, MaxLng: 360, Step: 90}) + if err != nil { + t.Fatalf("Rasterize within coverage: %v", err) + } + if got := out[0].Data[0]; got != 7 { + t.Errorf("u = %v, want 7", got) + } +} + +// TestRasterizeZeroFillsIndividualGaps preserves the documented per-cell +// behaviour: a partly-covered grid is still worth drawing, so gaps stay zero and +// the request succeeds. Only a grid with nothing in it is refused. +func TestRasterizeZeroFillsIndividualGaps(t *testing.T) { + out, err := Rasterize(northOnlyWind{}, Request{ + MinLat: -30, MaxLat: 30, MinLng: 0, MaxLng: 30, Step: 30, + }) + if err != nil { + t.Fatalf("Rasterize partly-covered grid: %v", err) + } + + // Row 0 is the northernmost latitude (+30), the last row is -30. + nx := out[0].Header.Nx + ny := out[0].Header.Ny + if got := out[0].Data[0]; got != 5 { + t.Errorf("northern cell u = %v, want 5", got) + } + if got := out[0].Data[(ny-1)*nx]; got != 0 { + t.Errorf("southern gap u = %v, want 0", got) + } +} diff --git a/pkg/rest/oas_client_gen.go b/pkg/rest/oas_client_gen.go index b7dad2f..c74265d 100644 --- a/pkg/rest/oas_client_gen.go +++ b/pkg/rest/oas_client_gen.go @@ -4,6 +4,7 @@ package rest import ( "context" + "io" "net/url" "strings" "time" @@ -239,7 +240,13 @@ func (c *Client) sendCancelDatasetJob(ctx context.Context, params CancelDatasetJ return res, errors.Wrap(err, "do request") } body := resp.Body - defer body.Close() + defer func() { + // Drain the body to EOF before closing, so the underlying + // connection can be reused by the Transport regardless of the + // response status code. See https://github.com/ogen-go/ogen/issues/1670. + _, _ = io.Copy(io.Discard, body) + _ = body.Close() + }() stage = "DecodeResponse" result, err := decodeCancelDatasetJobResponse(resp) @@ -331,7 +338,13 @@ func (c *Client) sendCancelPredictionJob(ctx context.Context, params CancelPredi return res, errors.Wrap(err, "do request") } body := resp.Body - defer body.Close() + defer func() { + // Drain the body to EOF before closing, so the underlying + // connection can be reused by the Transport regardless of the + // response status code. See https://github.com/ogen-go/ogen/issues/1670. + _, _ = io.Copy(io.Discard, body) + _ = body.Close() + }() stage = "DecodeResponse" result, err := decodeCancelPredictionJobResponse(resp) @@ -408,7 +421,13 @@ func (c *Client) sendCreatePredictionJob(ctx context.Context, request *Predictio return res, errors.Wrap(err, "do request") } body := resp.Body - defer body.Close() + defer func() { + // Drain the body to EOF before closing, so the underlying + // connection can be reused by the Transport regardless of the + // response status code. See https://github.com/ogen-go/ogen/issues/1670. + _, _ = io.Copy(io.Discard, body) + _ = body.Close() + }() stage = "DecodeResponse" result, err := decodeCreatePredictionJobResponse(resp) @@ -500,7 +519,13 @@ func (c *Client) sendDeleteDataset(ctx context.Context, params DeleteDatasetPara return res, errors.Wrap(err, "do request") } body := resp.Body - defer body.Close() + defer func() { + // Drain the body to EOF before closing, so the underlying + // connection can be reused by the Transport regardless of the + // response status code. See https://github.com/ogen-go/ogen/issues/1670. + _, _ = io.Copy(io.Discard, body) + _ = body.Close() + }() stage = "DecodeResponse" result, err := decodeDeleteDatasetResponse(resp) @@ -592,7 +617,13 @@ func (c *Client) sendGetDatasetJob(ctx context.Context, params GetDatasetJobPara return res, errors.Wrap(err, "do request") } body := resp.Body - defer body.Close() + defer func() { + // Drain the body to EOF before closing, so the underlying + // connection can be reused by the Transport regardless of the + // response status code. See https://github.com/ogen-go/ogen/issues/1670. + _, _ = io.Copy(io.Discard, body) + _ = body.Close() + }() stage = "DecodeResponse" result, err := decodeGetDatasetJobResponse(resp) @@ -684,7 +715,13 @@ func (c *Client) sendGetPredictionJob(ctx context.Context, params GetPredictionJ return res, errors.Wrap(err, "do request") } body := resp.Body - defer body.Close() + defer func() { + // Drain the body to EOF before closing, so the underlying + // connection can be reused by the Transport regardless of the + // response status code. See https://github.com/ogen-go/ogen/issues/1670. + _, _ = io.Copy(io.Discard, body) + _ = body.Close() + }() stage = "DecodeResponse" result, err := decodeGetPredictionJobResponse(resp) @@ -758,7 +795,13 @@ func (c *Client) sendGetServiceStatus(ctx context.Context) (res *StatusResponse, return res, errors.Wrap(err, "do request") } body := resp.Body - defer body.Close() + defer func() { + // Drain the body to EOF before closing, so the underlying + // connection can be reused by the Transport regardless of the + // response status code. See https://github.com/ogen-go/ogen/issues/1670. + _, _ = io.Copy(io.Discard, body) + _ = body.Close() + }() stage = "DecodeResponse" result, err := decodeGetServiceStatusResponse(resp) @@ -955,7 +998,13 @@ func (c *Client) sendGetWindField(ctx context.Context, params GetWindFieldParams return res, errors.Wrap(err, "do request") } body := resp.Body - defer body.Close() + defer func() { + // Drain the body to EOF before closing, so the underlying + // connection can be reused by the Transport regardless of the + // response status code. See https://github.com/ogen-go/ogen/issues/1670. + _, _ = io.Copy(io.Discard, body) + _ = body.Close() + }() stage = "DecodeResponse" result, err := decodeGetWindFieldResponse(resp) @@ -1029,7 +1078,13 @@ func (c *Client) sendGetWindMeta(ctx context.Context) (res *WindMeta, err error) return res, errors.Wrap(err, "do request") } body := resp.Body - defer body.Close() + defer func() { + // Drain the body to EOF before closing, so the underlying + // connection can be reused by the Transport regardless of the + // response status code. See https://github.com/ogen-go/ogen/issues/1670. + _, _ = io.Copy(io.Discard, body) + _ = body.Close() + }() stage = "DecodeResponse" result, err := decodeGetWindMetaResponse(resp) @@ -1103,7 +1158,13 @@ func (c *Client) sendListDatasetJobs(ctx context.Context) (res []DownloadJob, er return res, errors.Wrap(err, "do request") } body := resp.Body - defer body.Close() + defer func() { + // Drain the body to EOF before closing, so the underlying + // connection can be reused by the Transport regardless of the + // response status code. See https://github.com/ogen-go/ogen/issues/1670. + _, _ = io.Copy(io.Discard, body) + _ = body.Close() + }() stage = "DecodeResponse" result, err := decodeListDatasetJobsResponse(resp) @@ -1177,7 +1238,13 @@ func (c *Client) sendListDatasets(ctx context.Context) (res *DatasetList, err er return res, errors.Wrap(err, "do request") } body := resp.Body - defer body.Close() + defer func() { + // Drain the body to EOF before closing, so the underlying + // connection can be reused by the Transport regardless of the + // response status code. See https://github.com/ogen-go/ogen/issues/1670. + _, _ = io.Copy(io.Discard, body) + _ = body.Close() + }() stage = "DecodeResponse" result, err := decodeListDatasetsResponse(resp) @@ -1433,7 +1500,13 @@ func (c *Client) sendPerformPrediction(ctx context.Context, params PerformPredic return res, errors.Wrap(err, "do request") } body := resp.Body - defer body.Close() + defer func() { + // Drain the body to EOF before closing, so the underlying + // connection can be reused by the Transport regardless of the + // response status code. See https://github.com/ogen-go/ogen/issues/1670. + _, _ = io.Copy(io.Discard, body) + _ = body.Close() + }() stage = "DecodeResponse" result, err := decodePerformPredictionResponse(resp) @@ -1510,7 +1583,13 @@ func (c *Client) sendPerformPredictionV2(ctx context.Context, request *Predictio return res, errors.Wrap(err, "do request") } body := resp.Body - defer body.Close() + defer func() { + // Drain the body to EOF before closing, so the underlying + // connection can be reused by the Transport regardless of the + // response status code. See https://github.com/ogen-go/ogen/issues/1670. + _, _ = io.Copy(io.Discard, body) + _ = body.Close() + }() stage = "DecodeResponse" result, err := decodePerformPredictionV2Response(resp) @@ -1584,7 +1663,13 @@ func (c *Client) sendReadinessCheck(ctx context.Context) (res *ReadinessResponse return res, errors.Wrap(err, "do request") } body := resp.Body - defer body.Close() + defer func() { + // Drain the body to EOF before closing, so the underlying + // connection can be reused by the Transport regardless of the + // response status code. See https://github.com/ogen-go/ogen/issues/1670. + _, _ = io.Copy(io.Discard, body) + _ = body.Close() + }() stage = "DecodeResponse" result, err := decodeReadinessCheckResponse(resp) @@ -1661,7 +1746,13 @@ func (c *Client) sendTriggerDatasetDownload(ctx context.Context, request *Downlo return res, errors.Wrap(err, "do request") } body := resp.Body - defer body.Close() + defer func() { + // Drain the body to EOF before closing, so the underlying + // connection can be reused by the Transport regardless of the + // response status code. See https://github.com/ogen-go/ogen/issues/1670. + _, _ = io.Copy(io.Discard, body) + _ = body.Close() + }() stage = "DecodeResponse" result, err := decodeTriggerDatasetDownloadResponse(resp) diff --git a/pkg/rest/oas_handlers_gen.go b/pkg/rest/oas_handlers_gen.go index 4867c13..9acf62c 100644 --- a/pkg/rest/oas_handlers_gen.go +++ b/pkg/rest/oas_handlers_gen.go @@ -71,7 +71,7 @@ func (s *Server) handleCancelDatasetJobRequest(args [1]string, argsEscaped bool, if code != 0 { codeAttr := semconv.HTTPResponseStatusCode(code) attrs = append(attrs, codeAttr) - span.SetAttributes(codeAttr) + span.SetAttributes(attrs...) } attrOpt := metric.WithAttributes(attrs...) @@ -225,7 +225,7 @@ func (s *Server) handleCancelPredictionJobRequest(args [1]string, argsEscaped bo if code != 0 { codeAttr := semconv.HTTPResponseStatusCode(code) attrs = append(attrs, codeAttr) - span.SetAttributes(codeAttr) + span.SetAttributes(attrs...) } attrOpt := metric.WithAttributes(attrs...) @@ -379,7 +379,7 @@ func (s *Server) handleCreatePredictionJobRequest(args [0]string, argsEscaped bo if code != 0 { codeAttr := semconv.HTTPResponseStatusCode(code) attrs = append(attrs, codeAttr) - span.SetAttributes(codeAttr) + span.SetAttributes(attrs...) } attrOpt := metric.WithAttributes(attrs...) @@ -533,7 +533,7 @@ func (s *Server) handleDeleteDatasetRequest(args [1]string, argsEscaped bool, w if code != 0 { codeAttr := semconv.HTTPResponseStatusCode(code) attrs = append(attrs, codeAttr) - span.SetAttributes(codeAttr) + span.SetAttributes(attrs...) } attrOpt := metric.WithAttributes(attrs...) @@ -687,7 +687,7 @@ func (s *Server) handleGetDatasetJobRequest(args [1]string, argsEscaped bool, w if code != 0 { codeAttr := semconv.HTTPResponseStatusCode(code) attrs = append(attrs, codeAttr) - span.SetAttributes(codeAttr) + span.SetAttributes(attrs...) } attrOpt := metric.WithAttributes(attrs...) @@ -841,7 +841,7 @@ func (s *Server) handleGetPredictionJobRequest(args [1]string, argsEscaped bool, if code != 0 { codeAttr := semconv.HTTPResponseStatusCode(code) attrs = append(attrs, codeAttr) - span.SetAttributes(codeAttr) + span.SetAttributes(attrs...) } attrOpt := metric.WithAttributes(attrs...) @@ -995,7 +995,7 @@ func (s *Server) handleGetServiceStatusRequest(args [0]string, argsEscaped bool, if code != 0 { codeAttr := semconv.HTTPResponseStatusCode(code) attrs = append(attrs, codeAttr) - span.SetAttributes(codeAttr) + span.SetAttributes(attrs...) } attrOpt := metric.WithAttributes(attrs...) @@ -1130,7 +1130,7 @@ func (s *Server) handleGetWindFieldRequest(args [0]string, argsEscaped bool, w h if code != 0 { codeAttr := semconv.HTTPResponseStatusCode(code) attrs = append(attrs, codeAttr) - span.SetAttributes(codeAttr) + span.SetAttributes(attrs...) } attrOpt := metric.WithAttributes(attrs...) @@ -1308,7 +1308,7 @@ func (s *Server) handleGetWindMetaRequest(args [0]string, argsEscaped bool, w ht if code != 0 { codeAttr := semconv.HTTPResponseStatusCode(code) attrs = append(attrs, codeAttr) - span.SetAttributes(codeAttr) + span.SetAttributes(attrs...) } attrOpt := metric.WithAttributes(attrs...) @@ -1443,7 +1443,7 @@ func (s *Server) handleListDatasetJobsRequest(args [0]string, argsEscaped bool, if code != 0 { codeAttr := semconv.HTTPResponseStatusCode(code) attrs = append(attrs, codeAttr) - span.SetAttributes(codeAttr) + span.SetAttributes(attrs...) } attrOpt := metric.WithAttributes(attrs...) @@ -1578,7 +1578,7 @@ func (s *Server) handleListDatasetsRequest(args [0]string, argsEscaped bool, w h if code != 0 { codeAttr := semconv.HTTPResponseStatusCode(code) attrs = append(attrs, codeAttr) - span.SetAttributes(codeAttr) + span.SetAttributes(attrs...) } attrOpt := metric.WithAttributes(attrs...) @@ -1713,7 +1713,7 @@ func (s *Server) handlePerformPredictionRequest(args [0]string, argsEscaped bool if code != 0 { codeAttr := semconv.HTTPResponseStatusCode(code) attrs = append(attrs, codeAttr) - span.SetAttributes(codeAttr) + span.SetAttributes(attrs...) } attrOpt := metric.WithAttributes(attrs...) @@ -1907,7 +1907,7 @@ func (s *Server) handlePerformPredictionV2Request(args [0]string, argsEscaped bo if code != 0 { codeAttr := semconv.HTTPResponseStatusCode(code) attrs = append(attrs, codeAttr) - span.SetAttributes(codeAttr) + span.SetAttributes(attrs...) } attrOpt := metric.WithAttributes(attrs...) @@ -2061,7 +2061,7 @@ func (s *Server) handleReadinessCheckRequest(args [0]string, argsEscaped bool, w if code != 0 { codeAttr := semconv.HTTPResponseStatusCode(code) attrs = append(attrs, codeAttr) - span.SetAttributes(codeAttr) + span.SetAttributes(attrs...) } attrOpt := metric.WithAttributes(attrs...) @@ -2196,7 +2196,7 @@ func (s *Server) handleTriggerDatasetDownloadRequest(args [0]string, argsEscaped if code != 0 { codeAttr := semconv.HTTPResponseStatusCode(code) attrs = append(attrs, codeAttr) - span.SetAttributes(codeAttr) + span.SetAttributes(attrs...) } attrOpt := metric.WithAttributes(attrs...) diff --git a/pkg/rest/oas_json_gen.go b/pkg/rest/oas_json_gen.go index 21c5d84..bbd3acf 100644 --- a/pkg/rest/oas_json_gen.go +++ b/pkg/rest/oas_json_gen.go @@ -4417,6 +4417,12 @@ func (s *PredictionV2Request) encodeFields(e *jx.Encoder) { e.FieldStart("launch") s.Launch.Encode(e) } + { + if s.Dataset.Set { + e.FieldStart("dataset") + s.Dataset.Encode(e, json.EncodeDateTime) + } + } { if s.Direction.Set { e.FieldStart("direction") @@ -4449,12 +4455,13 @@ func (s *PredictionV2Request) encodeFields(e *jx.Encoder) { } } -var jsonFieldsNameOfPredictionV2Request = [5]string{ +var jsonFieldsNameOfPredictionV2Request = [6]string{ 0: "launch", - 1: "direction", - 2: "profile", - 3: "globals", - 4: "options", + 1: "dataset", + 2: "direction", + 3: "profile", + 4: "globals", + 5: "options", } // Decode decodes PredictionV2Request from json. @@ -4477,6 +4484,16 @@ func (s *PredictionV2Request) Decode(d *jx.Decoder) error { }(); err != nil { return errors.Wrap(err, "decode field \"launch\"") } + case "dataset": + if err := func() error { + s.Dataset.Reset() + if err := s.Dataset.Decode(d, json.DecodeDateTime); err != nil { + return err + } + return nil + }(); err != nil { + return errors.Wrap(err, "decode field \"dataset\"") + } case "direction": if err := func() error { s.Direction.Reset() @@ -4488,7 +4505,7 @@ func (s *PredictionV2Request) Decode(d *jx.Decoder) error { return errors.Wrap(err, "decode field \"direction\"") } case "profile": - requiredBitSet[0] |= 1 << 2 + requiredBitSet[0] |= 1 << 3 if err := func() error { s.Profile = make([]StageSpec, 0) if err := d.Arr(func(d *jx.Decoder) error { @@ -4542,7 +4559,7 @@ func (s *PredictionV2Request) Decode(d *jx.Decoder) error { // Validate required fields. var failures []validate.FieldError for i, mask := range [1]uint8{ - 0b00000101, + 0b00001001, } { if result := (requiredBitSet[i] & mask) ^ mask; result != 0 { // Mask only required fields and check equality to mask using XOR. diff --git a/pkg/rest/oas_response_encoders_gen.go b/pkg/rest/oas_response_encoders_gen.go index 33a690c..341c7fd 100644 --- a/pkg/rest/oas_response_encoders_gen.go +++ b/pkg/rest/oas_response_encoders_gen.go @@ -14,14 +14,12 @@ import ( func encodeCancelDatasetJobResponse(response *CancelDatasetJobNoContent, w http.ResponseWriter, span trace.Span) error { w.WriteHeader(204) - span.SetStatus(codes.Ok, http.StatusText(204)) return nil } func encodeCancelPredictionJobResponse(response *CancelPredictionJobNoContent, w http.ResponseWriter, span trace.Span) error { w.WriteHeader(204) - span.SetStatus(codes.Ok, http.StatusText(204)) return nil } @@ -29,7 +27,6 @@ func encodeCancelPredictionJobResponse(response *CancelPredictionJobNoContent, w func encodeCreatePredictionJobResponse(response *PredictionJob, w http.ResponseWriter, span trace.Span) error { w.Header().Set("Content-Type", "application/json; charset=utf-8") w.WriteHeader(202) - span.SetStatus(codes.Ok, http.StatusText(202)) e := new(jx.Encoder) response.Encode(e) @@ -42,7 +39,6 @@ func encodeCreatePredictionJobResponse(response *PredictionJob, w http.ResponseW func encodeDeleteDatasetResponse(response *DeleteDatasetNoContent, w http.ResponseWriter, span trace.Span) error { w.WriteHeader(204) - span.SetStatus(codes.Ok, http.StatusText(204)) return nil } @@ -50,7 +46,6 @@ func encodeDeleteDatasetResponse(response *DeleteDatasetNoContent, w http.Respon func encodeGetDatasetJobResponse(response *DownloadJob, w http.ResponseWriter, span trace.Span) error { w.Header().Set("Content-Type", "application/json; charset=utf-8") w.WriteHeader(200) - span.SetStatus(codes.Ok, http.StatusText(200)) e := new(jx.Encoder) response.Encode(e) @@ -64,7 +59,6 @@ func encodeGetDatasetJobResponse(response *DownloadJob, w http.ResponseWriter, s func encodeGetPredictionJobResponse(response *PredictionJob, w http.ResponseWriter, span trace.Span) error { w.Header().Set("Content-Type", "application/json; charset=utf-8") w.WriteHeader(200) - span.SetStatus(codes.Ok, http.StatusText(200)) e := new(jx.Encoder) response.Encode(e) @@ -78,7 +72,6 @@ func encodeGetPredictionJobResponse(response *PredictionJob, w http.ResponseWrit func encodeGetServiceStatusResponse(response *StatusResponse, w http.ResponseWriter, span trace.Span) error { w.Header().Set("Content-Type", "application/json; charset=utf-8") w.WriteHeader(200) - span.SetStatus(codes.Ok, http.StatusText(200)) e := new(jx.Encoder) response.Encode(e) @@ -92,7 +85,6 @@ func encodeGetServiceStatusResponse(response *StatusResponse, w http.ResponseWri func encodeGetWindFieldResponse(response []WindComponent, w http.ResponseWriter, span trace.Span) error { w.Header().Set("Content-Type", "application/json; charset=utf-8") w.WriteHeader(200) - span.SetStatus(codes.Ok, http.StatusText(200)) e := new(jx.Encoder) e.ArrStart() @@ -110,7 +102,6 @@ func encodeGetWindFieldResponse(response []WindComponent, w http.ResponseWriter, func encodeGetWindMetaResponse(response *WindMeta, w http.ResponseWriter, span trace.Span) error { w.Header().Set("Content-Type", "application/json; charset=utf-8") w.WriteHeader(200) - span.SetStatus(codes.Ok, http.StatusText(200)) e := new(jx.Encoder) response.Encode(e) @@ -124,7 +115,6 @@ func encodeGetWindMetaResponse(response *WindMeta, w http.ResponseWriter, span t func encodeListDatasetJobsResponse(response []DownloadJob, w http.ResponseWriter, span trace.Span) error { w.Header().Set("Content-Type", "application/json; charset=utf-8") w.WriteHeader(200) - span.SetStatus(codes.Ok, http.StatusText(200)) e := new(jx.Encoder) e.ArrStart() @@ -142,7 +132,6 @@ func encodeListDatasetJobsResponse(response []DownloadJob, w http.ResponseWriter func encodeListDatasetsResponse(response *DatasetList, w http.ResponseWriter, span trace.Span) error { w.Header().Set("Content-Type", "application/json; charset=utf-8") w.WriteHeader(200) - span.SetStatus(codes.Ok, http.StatusText(200)) e := new(jx.Encoder) response.Encode(e) @@ -156,7 +145,6 @@ func encodeListDatasetsResponse(response *DatasetList, w http.ResponseWriter, sp func encodePerformPredictionResponse(response *PredictionResponse, w http.ResponseWriter, span trace.Span) error { w.Header().Set("Content-Type", "application/json; charset=utf-8") w.WriteHeader(200) - span.SetStatus(codes.Ok, http.StatusText(200)) e := new(jx.Encoder) response.Encode(e) @@ -170,7 +158,6 @@ func encodePerformPredictionResponse(response *PredictionResponse, w http.Respon func encodePerformPredictionV2Response(response *PredictionV2Response, w http.ResponseWriter, span trace.Span) error { w.Header().Set("Content-Type", "application/json; charset=utf-8") w.WriteHeader(200) - span.SetStatus(codes.Ok, http.StatusText(200)) e := new(jx.Encoder) response.Encode(e) @@ -184,7 +171,6 @@ func encodePerformPredictionV2Response(response *PredictionV2Response, w http.Re func encodeReadinessCheckResponse(response *ReadinessResponse, w http.ResponseWriter, span trace.Span) error { w.Header().Set("Content-Type", "application/json; charset=utf-8") w.WriteHeader(200) - span.SetStatus(codes.Ok, http.StatusText(200)) e := new(jx.Encoder) response.Encode(e) @@ -198,7 +184,6 @@ func encodeReadinessCheckResponse(response *ReadinessResponse, w http.ResponseWr func encodeTriggerDatasetDownloadResponse(response *DownloadAccepted, w http.ResponseWriter, span trace.Span) error { w.Header().Set("Content-Type", "application/json; charset=utf-8") w.WriteHeader(202) - span.SetStatus(codes.Ok, http.StatusText(202)) e := new(jx.Encoder) response.Encode(e) @@ -217,10 +202,8 @@ func encodeErrorResponse(response *DefaultErrorStatusCode, w http.ResponseWriter code = http.StatusOK } w.WriteHeader(code) - if st := http.StatusText(code); code >= http.StatusBadRequest { - span.SetStatus(codes.Error, st) - } else { - span.SetStatus(codes.Ok, st) + if code >= http.StatusInternalServerError { + span.SetStatus(codes.Error, http.StatusText(code)) } e := new(jx.Encoder) diff --git a/pkg/rest/oas_schemas_gen.go b/pkg/rest/oas_schemas_gen.go index bdcd84a..75fbfba 100644 --- a/pkg/rest/oas_schemas_gen.go +++ b/pkg/rest/oas_schemas_gen.go @@ -2615,13 +2615,15 @@ func (s *PredictionResponseWarnings) init() PredictionResponseWarnings { return m } -// A profile-driven prediction. `profile` is an ordered chain of -// propagators; each integrates from where the previous ended. A stage's -// `constraints` decide when it ends and what happens next: stop the -// profile, hand off to `fallback_index`, or clip to the boundary. +// A profile-driven prediction. `profile` is an ordered chain of propagators; each integrates from +// where the previous ended. A stage's `constraints` decide when it ends and what happens next: stop +// the profile, hand off to `fallback_index`, or clip to the boundary. // Ref: #/components/schemas/PredictionV2Request type PredictionV2Request struct { Launch Launch `json:"launch"` + // Forecast run to predict from, given as its epoch. Defaults to the active dataset. The named run must + // be stored; a request for one that is not is rejected rather than answered from a different run. + Dataset OptDateTime `json:"dataset"` // Forward integrates launch→landing; reverse integrates backward in time. Direction OptPredictionV2RequestDirection `json:"direction"` Profile []StageSpec `json:"profile"` @@ -2635,6 +2637,11 @@ func (s *PredictionV2Request) GetLaunch() Launch { return s.Launch } +// GetDataset returns the value of Dataset. +func (s *PredictionV2Request) GetDataset() OptDateTime { + return s.Dataset +} + // GetDirection returns the value of Direction. func (s *PredictionV2Request) GetDirection() OptPredictionV2RequestDirection { return s.Direction @@ -2660,6 +2667,11 @@ func (s *PredictionV2Request) SetLaunch(val Launch) { s.Launch = val } +// SetDataset sets the value of Dataset. +func (s *PredictionV2Request) SetDataset(val OptDateTime) { + s.Dataset = val +} + // SetDirection sets the value of Direction. func (s *PredictionV2Request) SetDirection(val OptPredictionV2RequestDirection) { s.Direction = val