diff --git a/api/api.go b/api/api.go index cb487962..1ad876ed 100644 --- a/api/api.go +++ b/api/api.go @@ -55,6 +55,8 @@ func Routes() []*router.Route { router.NewRoute("POST", "/network/remove-client", handlers.RemoveNetworkClient), router.NewRoute("POST", "/network/remove-clients", handlers.RemoveNetworkClients), router.NewRoute("POST", "/network/provider-egress-location", handlers.ProviderEgressLocationSubmit), + router.NewRoute("GET", "/network/provider-egress-due", handlers.ProviderEgressLocationDue), + router.NewRoute("POST", "/network/provider-egress-attempt", handlers.ProviderEgressLocationAttempt), router.NewRoute("GET", "/network/clients", handlers.NetworkClients), router.NewRoute("GET", "/network/peers", handlers.NetworkPeers), router.NewRoute("GET", "/network/provider-locations", handlers.NetworkGetProviderLocations), diff --git a/api/handlers/provider_egress_location_handlers.go b/api/handlers/provider_egress_location_handlers.go index b843c05b..264da746 100644 --- a/api/handlers/provider_egress_location_handlers.go +++ b/api/handlers/provider_egress_location_handlers.go @@ -5,12 +5,14 @@ import ( "encoding/json" "io" "net/http" + "strconv" "sync" "github.com/urnetwork/glog" "github.com/urnetwork/server" "github.com/urnetwork/server/controller" + "github.com/urnetwork/server/model" ) // operatorSecretHeader carries the operator ingest secret. This endpoint is @@ -91,3 +93,137 @@ func ProviderEgressLocationSubmit(w http.ResponseWriter, r *http.Request) { glog.Infof("[pegl]could not write response. err = %s\n", err) } } + +// ProviderEgressLocationAttempt records that the operator's prober tried to +// probe a provider, whether or not the try produced a location. +// +// The prober reports a *failure* here; a success is reported by +// ProviderEgressLocationSubmit above, whose provider_egress_location row +// already defers the provider for the full staleness window. Reporting a +// success here as well is harmless -- the attempt backoff is far shorter than +// that window -- but redundant. +// +// This exists because ProviderEgressLocationDue would otherwise be starved by +// providers that can never be probed successfully: they never get an egress +// row, so they sort to the head of the queue on every poll forever. See +// model.GetProviderEgressLocationDue. +// +// Same auth as the two endpoints around it: operator-to-server, the shared +// secret header rather than a network jwt, fail-closed when the vault resource +// is missing. +func ProviderEgressLocationAttempt(w http.ResponseWriter, r *http.Request) { + secret := operatorIngestSecret() + provided := r.Header.Get(operatorSecretHeader) + if secret == "" || provided == "" || !hmac.Equal([]byte(secret), []byte(provided)) { + http.Error(w, "Unauthorized", http.StatusUnauthorized) + return + } + + body, err := io.ReadAll(io.LimitReader(r.Body, maxProviderEgressLocationBody+1)) + if err != nil { + http.Error(w, "Bad request", http.StatusBadRequest) + return + } + if len(body) > maxProviderEgressLocationBody { + http.Error(w, "Request too large", http.StatusRequestEntityTooLarge) + return + } + + var args controller.RecordProviderEgressProbeAttemptArgs + if err := json.Unmarshal(body, &args); err != nil { + http.Error(w, "Bad request", http.StatusBadRequest) + return + } + + result, err := controller.RecordProviderEgressProbeAttempt(r.Context(), &args) + if err != nil { + http.Error(w, err.Error(), http.StatusBadRequest) + return + } + + w.Header().Set("Content-Type", "application/json") + if err := json.NewEncoder(w).Encode(result); err != nil { + glog.Infof("[pegl]could not write response. err = %s\n", err) + } +} + +const ( + // defaultProviderEgressDueLimit is the batch size when the caller does not + // ask for one. + defaultProviderEgressDueLimit = 100 + // maxProviderEgressDueLimit bounds the batch size regardless of what the + // caller asks for, so one request cannot ask the database for the entire + // provider population. + maxProviderEgressDueLimit = 500 +) + +// providerEgressDueAge is how stale a stored probe must be before its provider +// is offered up for re-probing. It is deliberately shorter than +// model.ProviderEgressLocationMaxAge -- the age past which a stored location +// stops being trusted at all. If the two were equal, every location would lapse +// to the mmdb fallback at the exact moment it became due and stay lapsed until +// the prober worked its way around to it; at half the max age the prober has a +// full max-age/2 window to refresh a location before it expires. +const providerEgressDueAge = model.ProviderEgressLocationMaxAge / 2 + +// ProviderEgressLocationDueResult is the response body of +// ProviderEgressLocationDue. +type ProviderEgressLocationDueResult struct { + ClientIds []server.Id `json:"client_ids"` +} + +// ProviderEgressLocationDue tells the operator's prober which providers to +// probe next: those whose egress location has gone stale, and those that have +// never been probed at all, oldest first. +// +// This moves the probe schedule from the prober's memory into the database. +// The prober used to decide what to probe from an in-memory ttl cache, so a +// restart re-probed the whole population and nothing durable recorded what was +// actually due; observed_at already carries that information server-side, and +// this exposes it. +// +// A provider is skipped if it has a fresh success *or* a recent attempt. The +// second cutoff, ProviderEgressProbeAttemptBackoff, is much shorter than the +// first: a provider that failed to probe should be retried within hours, but +// must not be handed back on every poll, which is what would starve the rest of +// the queue (see ProviderEgressLocationAttempt above). +// +// Same auth as ProviderEgressLocationSubmit above: operator-to-server, the +// shared secret header rather than a network jwt, fail-closed when the vault +// resource is missing. +func ProviderEgressLocationDue(w http.ResponseWriter, r *http.Request) { + secret := operatorIngestSecret() + provided := r.Header.Get(operatorSecretHeader) + if secret == "" || provided == "" || !hmac.Equal([]byte(secret), []byte(provided)) { + http.Error(w, "Unauthorized", http.StatusUnauthorized) + return + } + + limit := defaultProviderEgressDueLimit + if raw := r.URL.Query().Get("limit"); raw != "" { + parsed, err := strconv.Atoi(raw) + if err != nil || parsed < 1 { + // not clamped up to 1: `limit=0` would come back as an empty list, + // which the prober cannot distinguish from "nothing is due" + http.Error(w, "Bad request", http.StatusBadRequest) + return + } + limit = min(parsed, maxProviderEgressDueLimit) + } + + // both cutoffs are computed here and passed as arguments; observed_at and + // attempt_at are naive timestamps holding utc, so comparing them to sql + // now() in the query would cast through the session timezone + now := server.NowUtc() + minObservedAt := now.Add(-providerEgressDueAge) + minAttemptAt := now.Add(-model.ProviderEgressProbeAttemptBackoff) + + result := &ProviderEgressLocationDueResult{ + ClientIds: model.GetProviderEgressLocationDue(r.Context(), minObservedAt, minAttemptAt, limit), + } + + w.Header().Set("Content-Type", "application/json") + if err := json.NewEncoder(w).Encode(result); err != nil { + glog.Infof("[pegl]could not write response. err = %s\n", err) + } +} diff --git a/api/handlers/provider_egress_location_handlers_test.go b/api/handlers/provider_egress_location_handlers_test.go index 4af00971..0f094ca0 100644 --- a/api/handlers/provider_egress_location_handlers_test.go +++ b/api/handlers/provider_egress_location_handlers_test.go @@ -2,14 +2,18 @@ package handlers import ( "bytes" + "context" "encoding/json" "net/http" "net/http/httptest" + "slices" "strings" "testing" + "time" "github.com/urnetwork/server" "github.com/urnetwork/server/controller" + "github.com/urnetwork/server/model" ) func TestProviderEgressLocationSubmitRejectsMissingSecret(t *testing.T) { @@ -148,3 +152,401 @@ func TestProviderEgressLocationSubmitReadsSecretFromVault(t *testing.T) { t.Fatalf("readOperatorIngestSecret() = %q, want %q", got, secret) } } + +func TestProviderEgressLocationDueRejectsMissingSecret(t *testing.T) { + req := httptest.NewRequest(http.MethodGet, "/network/provider-egress-due", nil) + w := httptest.NewRecorder() + + ProviderEgressLocationDue(w, req) + + if w.Code != http.StatusUnauthorized { + t.Fatalf("status = %d, want 401 when the operator secret header is absent", w.Code) + } +} + +func TestProviderEgressLocationDueRejectsWrongSecret(t *testing.T) { + req := httptest.NewRequest(http.MethodGet, "/network/provider-egress-due", nil) + req.Header.Set(operatorSecretHeader, "definitely-not-the-secret") + w := httptest.NewRecorder() + + ProviderEgressLocationDue(w, req) + + if w.Code != http.StatusUnauthorized { + t.Fatalf("status = %d, want 401 on a wrong operator secret", w.Code) + } +} + +// TestProviderEgressLocationDueRejectsAlteredSecret is the reject case with +// the vault *configured*, so the request gets past the secret == "" fail-closed +// short-circuit and hmac.Equal is what does the rejecting. +func TestProviderEgressLocationDueRejectsAlteredSecret(t *testing.T) { + const secret = "correct-operator-secret-0123456789" + const wrongSecret = "correct-operator-secret-0123456780" // last char changed + defer withStubOperatorIngestSecret(secret)() + + req := httptest.NewRequest(http.MethodGet, "/network/provider-egress-due", nil) + req.Header.Set(operatorSecretHeader, wrongSecret) + w := httptest.NewRecorder() + + ProviderEgressLocationDue(w, req) + + if w.Code != http.StatusUnauthorized { + t.Fatalf("status = %d, want 401 when the operator secret is configured but the request's secret is wrong", w.Code) + } +} + +// testing_connectDueProvider stands up a connected + valid provider holding a +// Public provide key and no probe result, i.e. a provider the due query must +// return. The caller runs model.UpdateClientLocationReliabilities afterward. +func testing_connectDueProvider( + t testing.TB, + ctx context.Context, + clientId server.Id, + locationId server.Id, + clientAddress string, +) { + model.Testing_CreateDevice(ctx, server.NewId(), server.NewId(), clientId, "", "") + + handlerId := model.CreateNetworkClientHandler(ctx) + connectionId, _, _, _, err := model.ConnectNetworkClient(ctx, clientId, clientAddress, handlerId) + if err != nil { + t.Fatalf("connect client: %s", err) + } + if err := model.SetConnectionLocation(ctx, connectionId, locationId, &model.ConnectionLocationScores{}); err != nil { + t.Fatalf("set connection location: %s", err) + } + model.SetProvide(ctx, clientId, map[model.ProvideMode][]byte{ + model.ProvideModePublic: []byte("provide-secret"), + }) +} + +// TestProviderEgressLocationDueAcceptsCorrectSecret proves the auth gate can +// ACCEPT. This is the test that gives the three reject tests above their +// meaning: without it, a handler whose entire body was replaced with an +// unconditional `http.Error(w, "Unauthorized", 401)` would still pass all +// three, because a suite that only ever asserts rejections cannot tell a +// working auth check from a broken-shut one. +// +// It deliberately asserts more than "not 401": a real, never-probed provider +// is stood up in the test database and must come back in the response body, so +// the test also fails if the handler clears auth but never reaches the model +// query or writes the wrong json shape. +func TestProviderEgressLocationDueAcceptsCorrectSecret(t *testing.T) { + t.Setenv("WARP_ENV", "local") + server.DefaultTestEnv().Run(t, func(t testing.TB) { + const secret = "correct-operator-secret-0123456789" + defer withStubOperatorIngestSecret(secret)() + + ctx := context.Background() + + city := &model.Location{ + LocationType: model.LocationTypeCity, + City: "Palo Alto", + Region: "California", + Country: "United States", + CountryCode: "us", + } + model.CreateLocation(ctx, city) + + due := server.NewId() + testing_connectDueProvider(t, ctx, due, city.LocationId, "0.0.0.1:0") + model.UpdateClientLocationReliabilities(ctx, server.NowUtc().Add(-time.Hour), server.NowUtc()) + + req := httptest.NewRequest(http.MethodGet, "/network/provider-egress-due?limit=10", nil) + req.Header.Set(operatorSecretHeader, secret) + w := httptest.NewRecorder() + + ProviderEgressLocationDue(w, req) + + if w.Code == http.StatusUnauthorized { + t.Fatalf("status = %d, want the correct secret to clear auth (not 401)", w.Code) + } + if w.Code != http.StatusOK { + t.Fatalf("status = %d, want 200; body = %s", w.Code, w.Body.String()) + } + + var result ProviderEgressLocationDueResult + if err := json.Unmarshal(w.Body.Bytes(), &result); err != nil { + t.Fatalf("decode body %q: %s", w.Body.String(), err) + } + if !slices.Contains(result.ClientIds, due) { + t.Fatalf("client_ids = %v, want it to contain the never-probed provider %s", result.ClientIds, due) + } + // the wire name the prober reads + if !strings.Contains(w.Body.String(), `"client_ids"`) { + t.Fatalf("body = %s, want a client_ids field", w.Body.String()) + } + }) +} + +// The prober asks for a batch; the server must not hand back more than asked. +func TestProviderEgressLocationDueHonoursLimit(t *testing.T) { + t.Setenv("WARP_ENV", "local") + server.DefaultTestEnv().Run(t, func(t testing.TB) { + const secret = "correct-operator-secret-0123456789" + defer withStubOperatorIngestSecret(secret)() + + ctx := context.Background() + + city := &model.Location{ + LocationType: model.LocationTypeCity, + City: "Palo Alto", + Region: "California", + Country: "United States", + CountryCode: "us", + } + model.CreateLocation(ctx, city) + + testing_connectDueProvider(t, ctx, server.NewId(), city.LocationId, "0.0.0.1:0") + testing_connectDueProvider(t, ctx, server.NewId(), city.LocationId, "0.0.0.2:0") + model.UpdateClientLocationReliabilities(ctx, server.NowUtc().Add(-time.Hour), server.NowUtc()) + + req := httptest.NewRequest(http.MethodGet, "/network/provider-egress-due?limit=1", nil) + req.Header.Set(operatorSecretHeader, secret) + w := httptest.NewRecorder() + + ProviderEgressLocationDue(w, req) + + if w.Code != http.StatusOK { + t.Fatalf("status = %d, want 200; body = %s", w.Code, w.Body.String()) + } + var result ProviderEgressLocationDueResult + if err := json.Unmarshal(w.Body.Bytes(), &result); err != nil { + t.Fatalf("decode body %q: %s", w.Body.String(), err) + } + if len(result.ClientIds) != 1 { + t.Fatalf("len(client_ids) = %d, want 1 for limit=1; body = %s", len(result.ClientIds), w.Body.String()) + } + }) +} + +// Every other due test in this file stands up never-probed providers, which +// come back regardless of what cutoff the handler computes -- so nothing here +// actually exercised providerEgressDueAge. This one does: a provider probed +// just now must be held back, and one probed past the cutoff must come through. +// Defeating the cutoff (dropping it, computing it in the wrong direction, +// comparing against sql now() through the session timezone) fails this. +func TestProviderEgressLocationDueHonoursStalenessCutoff(t *testing.T) { + t.Setenv("WARP_ENV", "local") + server.DefaultTestEnv().Run(t, func(t testing.TB) { + const secret = "correct-operator-secret-0123456789" + defer withStubOperatorIngestSecret(secret)() + + ctx := context.Background() + + city := &model.Location{ + LocationType: model.LocationTypeCity, + City: "Palo Alto", + Region: "California", + Country: "United States", + CountryCode: "us", + } + model.CreateLocation(ctx, city) + + fresh := server.NewId() + stale := server.NewId() + testing_connectDueProvider(t, ctx, fresh, city.LocationId, "0.0.0.1:0") + testing_connectDueProvider(t, ctx, stale, city.LocationId, "0.0.0.2:0") + model.UpdateClientLocationReliabilities(ctx, server.NowUtc().Add(-time.Hour), server.NowUtc()) + + now := server.NowUtc() + model.SetProviderEgressLocation(ctx, &model.ProviderEgressLocation{ + ClientId: fresh, LocationId: city.LocationId, + CountryCode: "us", ObservedAt: now, + }) + // comfortably past providerEgressDueAge, which is half + // model.ProviderEgressLocationMaxAge + model.SetProviderEgressLocation(ctx, &model.ProviderEgressLocation{ + ClientId: stale, LocationId: city.LocationId, + CountryCode: "us", ObservedAt: now.Add(-providerEgressDueAge - time.Hour), + }) + + req := httptest.NewRequest(http.MethodGet, "/network/provider-egress-due?limit=100", nil) + req.Header.Set(operatorSecretHeader, secret) + w := httptest.NewRecorder() + + ProviderEgressLocationDue(w, req) + + if w.Code != http.StatusOK { + t.Fatalf("status = %d, want 200; body = %s", w.Code, w.Body.String()) + } + var result ProviderEgressLocationDueResult + if err := json.Unmarshal(w.Body.Bytes(), &result); err != nil { + t.Fatalf("decode body %q: %s", w.Body.String(), err) + } + if slices.Contains(result.ClientIds, fresh) { + t.Fatalf("client_ids = %v, must not contain the just-probed provider %s", result.ClientIds, fresh) + } + if !slices.Contains(result.ClientIds, stale) { + t.Fatalf("client_ids = %v, must contain the provider probed past the cutoff %s", result.ClientIds, stale) + } + }) +} + +func TestProviderEgressLocationAttemptRejectsMissingSecret(t *testing.T) { + body, _ := json.Marshal(map[string]any{ + "client_id": "019f8835-158d-6fd8-e9dd-fd0e4c6d6792", + }) + req := httptest.NewRequest(http.MethodPost, "/network/provider-egress-attempt", bytes.NewReader(body)) + w := httptest.NewRecorder() + + ProviderEgressLocationAttempt(w, req) + + if w.Code != http.StatusUnauthorized { + t.Fatalf("status = %d, want 401 when the operator secret header is absent", w.Code) + } +} + +// The reject case with the vault *configured*, so the request gets past the +// secret == "" fail-closed short-circuit and hmac.Equal is what rejects. +func TestProviderEgressLocationAttemptRejectsAlteredSecret(t *testing.T) { + const secret = "correct-operator-secret-0123456789" + const wrongSecret = "correct-operator-secret-0123456780" // last char changed + defer withStubOperatorIngestSecret(secret)() + + body, _ := json.Marshal(map[string]any{ + "client_id": "019f8835-158d-6fd8-e9dd-fd0e4c6d6792", + }) + req := httptest.NewRequest(http.MethodPost, "/network/provider-egress-attempt", bytes.NewReader(body)) + req.Header.Set(operatorSecretHeader, wrongSecret) + w := httptest.NewRecorder() + + ProviderEgressLocationAttempt(w, req) + + if w.Code != http.StatusUnauthorized { + t.Fatalf("status = %d, want 401 when the operator secret is configured but the request's secret is wrong", w.Code) + } +} + +// The whole point of the attempt endpoint, end to end over http: a provider +// that has never been probed successfully is due; the prober reports that it +// tried and failed; the provider stops being due. Without that, a provider +// whose probes always fail sits at the head of the queue on every poll forever +// (observed_at IS NULL sorts first) and starves every provider behind it. +func TestProviderEgressLocationAttemptDefersProvider(t *testing.T) { + t.Setenv("WARP_ENV", "local") + server.DefaultTestEnv().Run(t, func(t testing.TB) { + const secret = "correct-operator-secret-0123456789" + defer withStubOperatorIngestSecret(secret)() + + ctx := context.Background() + + city := &model.Location{ + LocationType: model.LocationTypeCity, + City: "Palo Alto", + Region: "California", + Country: "United States", + CountryCode: "us", + } + model.CreateLocation(ctx, city) + + dead := server.NewId() + testing_connectDueProvider(t, ctx, dead, city.LocationId, "0.0.0.1:0") + model.UpdateClientLocationReliabilities(ctx, server.NowUtc().Add(-time.Hour), server.NowUtc()) + + if !slices.Contains(due(t, secret), dead) { + t.Fatalf("the never-probed provider %s must be due before any attempt is reported", dead) + } + + attemptBody, err := json.Marshal(controller.RecordProviderEgressProbeAttemptArgs{ + ClientId: dead, + ProbeFailure: "tunnel_failed", + }) + if err != nil { + t.Fatalf("marshal attempt: %s", err) + } + req := httptest.NewRequest(http.MethodPost, "/network/provider-egress-attempt", bytes.NewReader(attemptBody)) + req.Header.Set(operatorSecretHeader, secret) + w := httptest.NewRecorder() + + ProviderEgressLocationAttempt(w, req) + + if w.Code != http.StatusOK { + t.Fatalf("status = %d, want 200; body = %s", w.Code, w.Body.String()) + } + + attempt := model.GetProviderEgressProbeAttempt(ctx, dead) + if attempt == nil { + t.Fatal("expected the attempt to be recorded") + } + if attempt.ProbeFailure != "tunnel_failed" { + t.Fatalf("probe_failure = %q, want %q", attempt.ProbeFailure, "tunnel_failed") + } + + if slices.Contains(due(t, secret), dead) { + t.Fatalf("the provider %s must not be due again immediately after a failed attempt", dead) + } + }) +} + +// due drives the due endpoint over http and returns the batch. +func due(t testing.TB, secret string) []server.Id { + req := httptest.NewRequest(http.MethodGet, "/network/provider-egress-due?limit=100", nil) + req.Header.Set(operatorSecretHeader, secret) + w := httptest.NewRecorder() + + ProviderEgressLocationDue(w, req) + + if w.Code != http.StatusOK { + t.Fatalf("due: status = %d, want 200; body = %s", w.Code, w.Body.String()) + } + var result ProviderEgressLocationDueResult + if err := json.Unmarshal(w.Body.Bytes(), &result); err != nil { + t.Fatalf("due: decode body %q: %s", w.Body.String(), err) + } + return result.ClientIds +} + +// An unknown client id must be rejected rather than writing an attempt row +// keyed to a client that does not exist, which nothing would ever read. +func TestProviderEgressLocationAttemptRejectsUnknownClient(t *testing.T) { + t.Setenv("WARP_ENV", "local") + server.DefaultTestEnv().Run(t, func(t testing.TB) { + const secret = "correct-operator-secret-0123456789" + defer withStubOperatorIngestSecret(secret)() + + body, err := json.Marshal(controller.RecordProviderEgressProbeAttemptArgs{ + ClientId: server.NewId(), + ProbeFailure: "tunnel_failed", + }) + if err != nil { + t.Fatalf("marshal attempt: %s", err) + } + req := httptest.NewRequest(http.MethodPost, "/network/provider-egress-attempt", bytes.NewReader(body)) + req.Header.Set(operatorSecretHeader, secret) + w := httptest.NewRecorder() + + ProviderEgressLocationAttempt(w, req) + + if w.Code == http.StatusUnauthorized { + t.Fatalf("status = %d, want the correct secret to clear auth (not 401)", w.Code) + } + if w.Code != http.StatusBadRequest { + t.Fatalf("status = %d, want 400 for an unregistered client id; body = %s", w.Code, w.Body.String()) + } + if !strings.Contains(w.Body.String(), "Unknown client.") { + t.Fatalf("body = %q, want it to report the unknown client", w.Body.String()) + } + }) +} + +// A limit that is not a positive integer is a caller bug. Silently clamping it +// to 1 (or to the default) would answer a question the prober did not ask -- +// `limit=0` would come back as an empty list, indistinguishable from "nothing +// is due" -- so it is rejected instead. +func TestProviderEgressLocationDueRejectsBadLimit(t *testing.T) { + const secret = "correct-operator-secret-0123456789" + defer withStubOperatorIngestSecret(secret)() + + for _, raw := range []string{"0", "-1", "abc", "1.5"} { + req := httptest.NewRequest(http.MethodGet, "/network/provider-egress-due?limit="+raw, nil) + req.Header.Set(operatorSecretHeader, secret) + w := httptest.NewRecorder() + + ProviderEgressLocationDue(w, req) + + if w.Code != http.StatusBadRequest { + t.Fatalf("limit=%q: status = %d, want 400", raw, w.Code) + } + } +} diff --git a/controller/provider_egress_location_controller.go b/controller/provider_egress_location_controller.go index 169de94a..34ae3d81 100644 --- a/controller/provider_egress_location_controller.go +++ b/controller/provider_egress_location_controller.go @@ -110,23 +110,53 @@ func SubmitProviderEgressLocation( } } - // resolve to a canonical location row. city granularity only when the - // probe agreed on a city; otherwise country. - location := &model.Location{ - LocationType: model.LocationTypeCountry, - Country: country, - CountryCode: countryCode, - } + // resolve to a location row. City granularity only when the probe agreed + // on a city AND that city already exists in the location table. + // + // The probe MUST NOT define new cities or regions. model.CreateLocation + // dedupes a city on its exact location_name, so an unrecognised spelling + // does not fail -- it silently inserts a new permanent row into the shared + // `location` table and adds it to the search index. The three free + // geolocation sources the prober reaches consensus over demonstrably + // disagree on spelling ("Frankfurt am Main (Innenstadt I)" vs "Frankfurt am + // Main" for the same host, observed), and the consensus keeps the winning + // source's original display string -- so "Frankfurt am Main", "Frankfurt Am + // Main" and "Frankfurt/Main" would each become their own row. Those rows + // survive a code revert and there is no cleanup path. + // + // model.MatchExistingLocation therefore matches only, never creates, + // case-insensitively and ignoring punctuation/whitespace so the ordinary + // variants land on the row that is already there. When it does not resolve, + // this submission falls back to country granularity: country is the + // granularity this design treats as trustworthy anyway, and losing city + // precision for one probe is strictly better than permanently polluting a + // table shared with the provider list and the location search. + var location *model.Location if args.CityConfident { + location = model.MatchExistingLocation(ctx, countryCode, region, city) + } + + // city_confident records the granularity of the row actually stored, not + // what the probe claimed. The schema's documented invariant is that + // location_id is a city row exactly when city_confident is set (see the + // provider_egress_location migration), and a city-confident probe whose + // city did not resolve is stored at country granularity. + cityConfident := location != nil + + if location == nil { + // country granularity. This still goes through CreateLocation: a + // country row is keyed on country_code, so a variant *name* can never + // produce a second row for the same country the way a variant city name + // can -- the pollution this guards against is not reachable here. The + // country row is also the whole point of the fallback, so a probe from + // a country not yet in the table must not be dropped. location = &model.Location{ - LocationType: model.LocationTypeCity, - City: city, - Region: region, + LocationType: model.LocationTypeCountry, Country: country, CountryCode: countryCode, } + model.CreateLocation(ctx, location) } - model.CreateLocation(ctx, location) model.SetProviderEgressLocation(ctx, &model.ProviderEgressLocation{ ClientId: args.ClientId, @@ -137,9 +167,61 @@ func SubmitProviderEgressLocation( Hosting: args.Hosting, Proxy: args.Proxy, Mobile: args.Mobile, - CityConfident: args.CityConfident, + CityConfident: cityConfident, ObservedAt: args.ObservedAt, }) return &SubmitProviderEgressLocationResult{LocationId: location.LocationId}, nil } + +// maxProbeFailureLen bounds the failure class as submitted: +// provider_egress_probe_attempt.probe_failure is a varchar(64), and rejecting +// an over-long value with a clear error beats letting the insert panic on a +// Postgres "value too long" error and spin in the retry loop. +const maxProbeFailureLen = 64 + +type RecordProviderEgressProbeAttemptArgs struct { + ClientId server.Id `json:"client_id"` + // ProbeFailure is "" when the attempt succeeded, otherwise a short failure + // class (`contract_failed`, `tunnel_failed`, `no_consensus`, ...). + ProbeFailure string `json:"probe_failure,omitempty"` +} + +type RecordProviderEgressProbeAttemptResult struct { + AttemptAt time.Time `json:"attempt_at"` +} + +// RecordProviderEgressProbeAttempt records that the prober tried this provider. +// A failed attempt defers the provider from the due queue for +// ProviderEgressProbeAttemptBackoff, exactly as a successful probe defers it +// for the (much longer) staleness window -- without this, a provider that +// always fails to probe never gets a provider_egress_location row and so stays +// permanently at the head of the queue, starving every other provider. See +// model.GetProviderEgressLocationDue. +// +// The attempt is timestamped by the server, not the prober: the prober is +// reporting something it just did, and a prober whose clock ran fast could +// otherwise defer a provider far past the backoff window. +func RecordProviderEgressProbeAttempt( + ctx context.Context, + args *RecordProviderEgressProbeAttemptArgs, +) (*RecordProviderEgressProbeAttemptResult, error) { + if maxProbeFailureLen < len(args.ProbeFailure) { + return nil, fmt.Errorf("Probe failure class is too long.") + } + // same check as SubmitProviderEgressLocation: without it a typo'd or stale + // client id writes a row keyed to a client that does not exist, which + // nothing ever reads and only the sweep ever removes. + if networkId := model.GetNetworkClientNetwork(ctx, args.ClientId); networkId == nil { + return nil, fmt.Errorf("Unknown client.") + } + + attemptAt := server.NowUtc() + model.SetProviderEgressProbeAttempt(ctx, &model.ProviderEgressProbeAttempt{ + ClientId: args.ClientId, + AttemptAt: attemptAt, + ProbeFailure: args.ProbeFailure, + }) + + return &RecordProviderEgressProbeAttemptResult{AttemptAt: attemptAt}, nil +} diff --git a/controller/provider_egress_location_controller_test.go b/controller/provider_egress_location_controller_test.go index 25a04c6d..f3009046 100644 --- a/controller/provider_egress_location_controller_test.go +++ b/controller/provider_egress_location_controller_test.go @@ -106,6 +106,18 @@ func TestSubmitProviderEgressLocationCityConfidentStoresCity(t *testing.T) { clientId := server.NewId() model.Testing_CreateDevice(ctx, networkId, server.NewId(), clientId, "", "") + // the ingest path resolves against locations that ALREADY exist and + // never creates one, so the city has to be in the table first -- as it + // would be from the mmdb import + denver := &model.Location{ + LocationType: model.LocationTypeCity, + City: "Denver", + Region: "Colorado", + Country: "United States", + CountryCode: "us", + } + model.CreateLocation(ctx, denver) + _, err := SubmitProviderEgressLocation(ctx, &SubmitProviderEgressLocationArgs{ ClientId: clientId, CountryCode: "us", @@ -124,7 +136,9 @@ func TestSubmitProviderEgressLocationCityConfidentStoresCity(t *testing.T) { } assert.Equal(t, stored.CityConfident, true) - // the resolved location must be the city-granular row + // the resolved location must be the city-granular row that already + // existed, not a new one + assert.Equal(t, stored.LocationId, denver.LocationId) loc := model.GetLocation(ctx, stored.LocationId) if loc == nil { t.Fatal("expected the resolved location row to exist") @@ -293,3 +307,160 @@ func TestSubmitProviderEgressLocationRejectsStaleObservedAt(t *testing.T) { } }) } + +// testing_countLocations is the whole point of the two tests below: the +// `location` table is shared with the provider list and the location search, +// its rows are permanent, and nothing cleans up a bad one. An ingest endpoint +// that can add to it is an endpoint that can corrupt it from outside. +func testing_countLocations(ctx context.Context) int64 { + var count int64 + server.Db(ctx, func(conn server.PgConn) { + result, err := conn.Query(ctx, `SELECT COUNT(*) FROM location`) + server.WithPgResult(result, err, func() { + if result.Next() { + server.Raise(result.Scan(&count)) + } + }) + }) + return count +} + +// A city-confident submission whose city is not already in the location table +// must fall back to country granularity and must NOT create a location row. +// +// model.CreateLocation dedupes a city on its exact location_name, so before +// this fix an unrecognised spelling did not fail -- it silently inserted a new +// permanent row and indexed it for search. The prober's consensus stores the +// winning source's original display string and the three free geolocation +// sources demonstrably disagree on spelling, so "Frankfurt am Main", +// "Frankfurt Am Main" and "Frankfurt/Main" would each have become their own +// row. Those rows outlive a code revert and there is no cleanup path. +// +// Reverting the MatchExistingLocation call in SubmitProviderEgressLocation must +// fail this test: the row count goes up and the stored location is a city. +func TestSubmitProviderEgressLocationUnknownCityDoesNotCreateALocation(t *testing.T) { + server.DefaultTestEnv().Run(t, func(t testing.TB) { + ctx := context.Background() + + networkId := server.NewId() + clientId := server.NewId() + model.Testing_CreateDevice(ctx, networkId, server.NewId(), clientId, "", "") + + // Germany, and one real German city, already exist -- as they would + // from the mmdb import. The submission below names a DIFFERENT city + // that has never been seen. + model.CreateLocation(ctx, &model.Location{ + LocationType: model.LocationTypeCity, + City: "Frankfurt am Main", + Region: "Hesse", + Country: "Germany", + CountryCode: "de", + }) + + before := testing_countLocations(ctx) + + res, err := SubmitProviderEgressLocation(ctx, &SubmitProviderEgressLocationArgs{ + ClientId: clientId, + CountryCode: "de", + Country: "Germany", + Region: "Hesse", + City: "Kleinstadt Nirgendwo", + CountryConfident: true, + CityConfident: true, + ObservedAt: server.NowUtc(), + }) + assert.Equal(t, err, nil) + + // nothing was added to the shared table + after := testing_countLocations(ctx) + if after != before { + t.Errorf("location row count went from %d to %d; an unmatched city must not create a permanent row in the shared location table", before, after) + } + + // and the submission was stored at country granularity instead + stored := model.GetProviderEgressLocation(ctx, clientId) + if stored == nil { + t.Fatal("expected the submission to be stored") + } + loc := model.GetLocation(ctx, stored.LocationId) + if loc == nil { + t.Fatal("expected the resolved location row to exist") + } + if loc.LocationType != model.LocationTypeCountry { + t.Errorf("stored location_type = %q, want %q: an unmatched city must fall back to country granularity", loc.LocationType, model.LocationTypeCountry) + } + assert.Equal(t, res.LocationId, stored.LocationId) + + // city_confident tracks the granularity actually stored, so the row + // stays internally consistent: location_id is a city row exactly when + // city_confident is set + if stored.CityConfident { + t.Error("city_confident must be false when the submission was stored at country granularity") + } + }) +} + +// The variants that matter are trivial: the same city spelled with different +// case, punctuation or spacing. Those must resolve to the row that is already +// there -- discarding them to country would throw away real precision for no +// reason, and creating a row for each is the bug this guards against. +func TestSubmitProviderEgressLocationMatchesCitySpellingVariant(t *testing.T) { + server.DefaultTestEnv().Run(t, func(t testing.TB) { + ctx := context.Background() + + networkId := server.NewId() + model.Testing_CreateDevice(ctx, networkId, server.NewId(), server.NewId(), "", "") + + frankfurt := &model.Location{ + LocationType: model.LocationTypeCity, + City: "Frankfurt am Main", + Region: "Hesse", + Country: "Germany", + CountryCode: "de", + } + model.CreateLocation(ctx, frankfurt) + + before := testing_countLocations(ctx) + + // each of these is a real disagreement between the geolocation sources + // over the same place + variants := []struct{ region, city string }{ + {"Hesse", "Frankfurt am Main"}, // exact + {"Hesse", "Frankfurt Am Main"}, // case + {"hesse", "FRANKFURT AM MAIN"}, // case, both levels + {"Hesse", "Frankfurt-am-Main"}, // punctuation + {"Hesse", " Frankfurt am Main "}, + } + for _, variant := range variants { + clientId := server.NewId() + model.Testing_CreateDevice(ctx, networkId, server.NewId(), clientId, "", "") + + _, err := SubmitProviderEgressLocation(ctx, &SubmitProviderEgressLocationArgs{ + ClientId: clientId, + CountryCode: "DE", + Country: "Germany", + Region: variant.region, + City: variant.city, + CountryConfident: true, + CityConfident: true, + ObservedAt: server.NowUtc(), + }) + assert.Equal(t, err, nil) + + stored := model.GetProviderEgressLocation(ctx, clientId) + if stored == nil { + t.Fatalf("%q: expected the submission to be stored", variant.city) + } + if stored.LocationId != frankfurt.LocationId { + t.Errorf("%q resolved to %s, want the existing Frankfurt row %s", variant.city, stored.LocationId, frankfurt.LocationId) + } + if !stored.CityConfident { + t.Errorf("%q: city_confident must stay set when the city resolved", variant.city) + } + } + + if after := testing_countLocations(ctx); after != before { + t.Errorf("location row count went from %d to %d; spelling variants must reuse the existing row, not add new ones", before, after) + } + }) +} diff --git a/db_migrations.go b/db_migrations.go index be206cee..c1bbe401 100644 --- a/db_migrations.go +++ b/db_migrations.go @@ -4523,4 +4523,67 @@ var migrations = []any{ CREATE INDEX IF NOT EXISTS network_client_top_level_contract_time ON network_client (contract_time) WHERE (active = true AND source_client_id IS NULL AND contract_time IS NOT NULL) `), + + // provider egress probe attempts: when the prober last *tried* a provider, + // successful or not, and how the try failed. + // + // This cannot live on provider_egress_location, because the case it exists + // to handle is precisely a provider that has no row there. A provider that + // connects, holds a Public provide key and fails every probe (firewalled + // egress, dead upstream) never gets an egress row, so its observed_at stays + // NULL, so it sorts to the head of the due queue forever. Enough of them and + // every batch the prober asks for is the same set of permanently-dead + // providers, and no healthy provider's location is ever refreshed -- while + // the endpoint keeps returning a full, plausible-looking batch. + // GetProviderEgressLocationDue defers on a recent attempt as well as a fresh + // success, which needs somewhere to record the attempt. + // + // Pulled forward from the P2 verdict model + // (docs/superpowers/specs/2026-07-25-enforced-provider-geo-probing-design.md, + // probe_attempt_at / probe_failure) because the P1 schedule cannot function + // without it. Deliberately only the two columns the schedule reads, not the + // rest of that model. + newSqlMigration(` + CREATE TABLE IF NOT EXISTS provider_egress_probe_attempt ( + client_id uuid NOT NULL PRIMARY KEY, + attempt_at timestamp NOT NULL, + probe_failure varchar(64) NOT NULL DEFAULT '', + update_time timestamp NOT NULL + ) + `), + + // serves the sweep in RemoveExpiredProviderEgressProbeAttempts. The due + // query reaches this table by primary key through the left join, so it + // needs no index of its own. + newSqlMigration(` + CREATE INDEX IF NOT EXISTS provider_egress_probe_attempt_attempt_at + ON provider_egress_probe_attempt (attempt_at) + `), + + // serves the stale-but-probed pass of GetProviderEgressLocationDue, which + // drives from provider_egress_location with `observed_at < $n ORDER BY + // observed_at, client_id LIMIT $m`. With client_id in the index the + // predicate and the whole ORDER BY -- tie-break included -- are one ordered + // index scan that stops when the batch is full: no sort, and no heap visit + // to resolve the tie. The pre-existing (observed_at) index alone leaves the + // client_id tie-break to a sort. + // + // The other pass (never-probed) needs no new index: it is an anti-join over + // network_client_location_reliability ordered by client_id, which the + // existing (valid, connected, client_id) index already serves as an ordered + // scan, and both anti-joins plus the provide_key EXISTS are primary-key + // probes. + // + // This supersedes provider_egress_location_observed_at, which is now a + // prefix of it -- including for the RemoveExpiredProviderEgressLocations + // sweep. The redundant index is left in place deliberately: dropping it is a + // separate decision with its own (small) risk, and this migration is meant + // to be purely additive. + // + // Appended, never inserted: migrations here apply by slice index, so + // editing or reordering an already-applied entry corrupts live databases. + newSqlMigration(` + CREATE INDEX IF NOT EXISTS provider_egress_location_observed_at_client_id + ON provider_egress_location (observed_at, client_id) + `), } diff --git a/model/provider_egress_location_model.go b/model/provider_egress_location_model.go index 240bb8b4..88ee8603 100644 --- a/model/provider_egress_location_model.go +++ b/model/provider_egress_location_model.go @@ -4,6 +4,7 @@ import ( "context" "strings" "time" + "unicode" "github.com/urnetwork/server" ) @@ -13,6 +14,16 @@ import ( // mmdb lookup on the observed control ip. const ProviderEgressLocationMaxAge = 7 * 24 * time.Hour +// ProviderEgressProbeAttemptBackoff is how long a probe *attempt* defers a +// provider from being offered up again, whether or not the attempt succeeded. +// +// It is much shorter than the staleness window a successful probe buys +// (providerEgressDueAge in api/handlers, half ProviderEgressLocationMaxAge): a +// provider that fails to probe should be retried periodically -- the fault may +// be transient -- just not on every single poll, which is what starves the rest +// of the queue. +const ProviderEgressProbeAttemptBackoff = 6 * time.Hour + // ProviderEgressLocation is a provider location learned by probing the // provider's own egress, rather than by looking up its control-connection ip. type ProviderEgressLocation struct { @@ -86,6 +97,87 @@ func SetProviderEgressLocation(ctx context.Context, e *ProviderEgressLocation) { }) } +// ProviderEgressProbeAttempt records that the prober tried a provider, whether +// or not the try produced a location. +// +// A provider that has never been probed successfully has no +// ProviderEgressLocation row at all, so an attempt cannot be recorded there -- +// see the provider_egress_probe_attempt migration for why that matters. +// ProbeFailure is "" for a successful attempt, otherwise a short failure class +// (`tunnel_failed`, `no_consensus`, ...). +type ProviderEgressProbeAttempt struct { + ClientId server.Id + AttemptAt time.Time + ProbeFailure string + UpdateTime time.Time +} + +// SetProviderEgressProbeAttempt upserts the last probe attempt for a provider. +// +// Like SetProviderEgressLocation the upsert is monotonic in its timestamp: a +// replayed or out-of-order report older than what is already stored is dropped +// rather than moving the provider's last-attempt time backwards, which would +// hand it back to the prober early. +func SetProviderEgressProbeAttempt(ctx context.Context, a *ProviderEgressProbeAttempt) { + server.Tx(ctx, func(tx server.PgTx) { + server.RaisePgResult(tx.Exec( + ctx, + ` + INSERT INTO provider_egress_probe_attempt ( + client_id, + attempt_at, + probe_failure, + update_time + ) + VALUES ($1, $2, $3, $4) + ON CONFLICT (client_id) DO UPDATE + SET + attempt_at = $2, + probe_failure = $3, + update_time = $4 + WHERE provider_egress_probe_attempt.attempt_at < EXCLUDED.attempt_at + `, + a.ClientId, + a.AttemptAt.UTC(), + a.ProbeFailure, + server.NowUtc(), + )) + }) +} + +// GetProviderEgressProbeAttempt returns the last recorded probe attempt for a +// provider, or nil. +func GetProviderEgressProbeAttempt(ctx context.Context, clientId server.Id) *ProviderEgressProbeAttempt { + var a *ProviderEgressProbeAttempt + server.Db(ctx, func(conn server.PgConn) { + result, err := conn.Query( + ctx, + ` + SELECT + client_id, + attempt_at, + probe_failure, + update_time + FROM provider_egress_probe_attempt + WHERE client_id = $1 + `, + clientId, + ) + server.WithPgResult(result, err, func() { + if result.Next() { + a = &ProviderEgressProbeAttempt{} + server.Raise(result.Scan( + &a.ClientId, + &a.AttemptAt, + &a.ProbeFailure, + &a.UpdateTime, + )) + } + }) + }) + return a +} + // GetProviderEgressLocation returns the stored location for a provider, or nil. func GetProviderEgressLocation(ctx context.Context, clientId server.Id) *ProviderEgressLocation { var e *ProviderEgressLocation @@ -274,6 +366,459 @@ func GetLocation(ctx context.Context, locationId server.Id) *Location { return loc } +// normalizeLocationName folds a location name to a comparison key: lowercased, +// with every rune that is not a letter or a digit dropped. So +// "Frankfurt am Main", "Frankfurt Am Main" and "FRANKFURT AM MAIN" all fold to +// "frankfurtammain" and match the one row that already exists. +// +// This is deliberately a comparison key only -- it is never stored, and never +// used to build a location_name. It exists so a trivial spelling variant from a +// geolocation source resolves to the existing row instead of being treated as a +// different place. +// +// Punctuation is dropped rather than mapped to a space because the disagreement +// is over whether the separator exists at all ("Washington, D.C." vs +// "Washington DC"). Note this deliberately does not fold "Frankfurt/Main" onto +// "Frankfurt am Main": dropping the separator gives "frankfurtmain" != +// "frankfurtammain", so that one falls back to country granularity rather than +// matching the wrong row. Falling back is the safe outcome; guessing is not. +// Stdlib only, by design -- a transliteration/fuzzy-match dependency is a large +// amount of new behaviour to take on for an ingest path whose failure mode is +// already "use the country". +func normalizeLocationName(name string) string { + var b strings.Builder + b.Grow(len(name)) + for _, r := range strings.ToLower(name) { + if unicode.IsLetter(r) || unicode.IsDigit(r) { + b.WriteRune(r) + } + } + return b.String() +} + +// matchLocationNameInTx returns the location_id of the row in `candidates` +// whose location_name matches `name`, preferring an exact match and falling +// back to a normalized one (see normalizeLocationName), or nil for no match. +// Candidates must already be ordered deterministically by the caller so that +// two rows folding to the same key always resolve the same way. +func matchLocationName(name string, candidateIds []server.Id, candidateNames []string) *server.Id { + for i, candidateName := range candidateNames { + if candidateName == name { + return &candidateIds[i] + } + } + normalized := normalizeLocationName(name) + if normalized == "" { + // nothing comparable survives folding (e.g. a name of only + // punctuation); an empty key would match any other such row + return nil + } + for i, candidateName := range candidateNames { + if normalizeLocationName(candidateName) == normalized { + return &candidateIds[i] + } + } + return nil +} + +// MatchExistingLocation resolves (countryCode, region, city) against location +// rows that ALREADY EXIST and returns the city-granular row, or nil if any +// level of the hierarchy does not resolve. It never inserts anything. +// +// This is the resolver the provider egress ingest path uses instead of +// CreateLocation. CreateLocation deduplicates a city on its exact +// location_name, so an unrecognised spelling does not fail -- it silently +// creates a new, permanent row in the shared `location` table and indexes it +// for search. A geolocation probe has no business defining the world's cities: +// the three free sources the prober reaches consensus over demonstrably +// disagree on spelling (we observed "Frankfurt am Main (Innenstadt I)" against +// "Frankfurt am Main" for one host), and the consensus stores the winning +// source's original display string. Each variant would become its own row, +// those rows outlive a code revert, and there is no cleanup path. +// +// Matching is case-insensitive and ignores punctuation and whitespace +// differences, so the ordinary variants resolve to the row that is already +// there. When nothing resolves the caller falls back to country granularity -- +// see SubmitProviderEgressLocation. Falling back loses precision for one +// submission; creating a row corrupts shared data permanently. +// +// Each level tries an exact, fully-indexed match first (the common case: the +// winning source usually spells it the way the mmdb import did) and only scans +// the level's candidates when that misses. +func MatchExistingLocation( + ctx context.Context, + countryCode string, + region string, + city string, +) *Location { + countryCode = strings.ToLower(strings.TrimSpace(countryCode)) + region = strings.TrimSpace(region) + city = strings.TrimSpace(city) + if countryCode == "" || region == "" || city == "" { + return nil + } + + var match *Location + server.Db(ctx, func(conn server.PgConn) { + // country: keyed on country_code alone, exactly as CreateLocation + // dedupes it, so there is no name to match here + var countryLocationId server.Id + var countryName string + found := false + result, err := conn.Query( + ctx, + ` + SELECT location_id, location_name + FROM location + WHERE location_type = $1 AND country_code = $2 + ORDER BY location_id + LIMIT 1 + `, + LocationTypeCountry, + countryCode, + ) + server.WithPgResult(result, err, func() { + if result.Next() { + server.Raise(result.Scan(&countryLocationId, &countryName)) + found = true + } + }) + if !found { + return + } + + // region, within that country + regionLocationId := matchChildLocation( + ctx, + conn, + LocationTypeRegion, + countryCode, + region, + ` + SELECT location_id, location_name + FROM location + WHERE + location_type = $1 AND + country_code = $2 AND + location_name = $3 AND + country_location_id = $4 + `, + ` + SELECT location_id, location_name + FROM location + WHERE + location_type = $1 AND + country_code = $2 AND + country_location_id = $3 + ORDER BY location_id + `, + []any{countryLocationId}, + ) + if regionLocationId == nil { + return + } + + // city, within that region + cityLocationId := matchChildLocation( + ctx, + conn, + LocationTypeCity, + countryCode, + city, + ` + SELECT location_id, location_name + FROM location + WHERE + location_type = $1 AND + country_code = $2 AND + location_name = $3 AND + region_location_id = $4 AND + country_location_id = $5 + `, + ` + SELECT location_id, location_name + FROM location + WHERE + location_type = $1 AND + country_code = $2 AND + region_location_id = $3 AND + country_location_id = $4 + ORDER BY location_id + `, + []any{*regionLocationId, countryLocationId}, + ) + if cityLocationId == nil { + return + } + + match = &Location{ + LocationType: LocationTypeCity, + City: city, + Region: region, + Country: countryName, + CountryCode: countryCode, + LocationId: *cityLocationId, + CityLocationId: *cityLocationId, + RegionLocationId: *regionLocationId, + CountryLocationId: countryLocationId, + } + }) + return match +} + +// matchChildLocation runs the exact-match query first and only falls back to +// scanning the level's candidates when it misses. `parents` are the parent +// location ids the two queries scope on: the exact query binds them after +// (location_type, country_code, name), the candidate query after +// (location_type, country_code). +func matchChildLocation( + ctx context.Context, + conn server.PgConn, + locationType LocationType, + countryCode string, + name string, + exactSql string, + candidatesSql string, + parents []any, +) *server.Id { + exactArgs := append([]any{locationType, countryCode, name}, parents...) + var exactId *server.Id + result, err := conn.Query(ctx, exactSql, exactArgs...) + server.WithPgResult(result, err, func() { + if result.Next() { + var locationId server.Id + var locationName string + server.Raise(result.Scan(&locationId, &locationName)) + exactId = &locationId + } + }) + if exactId != nil { + return exactId + } + + candidateArgs := append([]any{locationType, countryCode}, parents...) + candidateIds := []server.Id{} + candidateNames := []string{} + result, err = conn.Query(ctx, candidatesSql, candidateArgs...) + server.WithPgResult(result, err, func() { + for result.Next() { + var locationId server.Id + var locationName string + server.Raise(result.Scan(&locationId, &locationName)) + candidateIds = append(candidateIds, locationId) + candidateNames = append(candidateNames, locationName) + } + }) + return matchLocationName(name, candidateIds, candidateNames) +} + +// GetProviderEgressLocationDue returns the client ids of providers whose +// egress location is due for a probe: no fresh success (newest probe older than +// minObservedAt, or never probed) *and* no recent attempt (last attempt older +// than minAttemptAt, or never attempted). Oldest first, so the +// longest-unprobed are handed out first, capped at limit. +// +// This is the durable replacement for the prober's in-memory ttl cache: the +// schedule lives in the database, so a prober restart resumes where it left +// off instead of re-probing everything. +// +// Three things about the shape of this query matter. +// +// First, candidates are sourced from the live provider population +// (network_client_location_reliability, connected + valid) and the egress row +// is LEFT JOINed on. The dominant case by far is a provider that has *never* +// been probed and therefore has no provider_egress_location row at all; +// selecting from provider_egress_location would return exactly the providers +// that least need probing and none of the ones that most do. +// +// Second, only providers holding a Public provide key are returned. Probing +// tunnels through the provider itself, which means opening a contract from +// outside the provider's own network -- something a provider without a Public +// key refuses. Offering one to the prober would burn a probe slot on a +// guaranteed failure. This is the same filter UpdateClientLocations and +// UpdateClientScores apply (network_client_location_model.go). +// +// Third, a recent *attempt* defers a provider the same way a recent success +// does. Without that, a provider that connects and holds a Public provide key +// but always fails to probe -- for any reason other than the missing Public key +// screened for above -- never gets an egress row, so its observed_at stays +// NULL, so it sorts ahead of every stale-but-refreshable provider on every +// single poll, forever. Enough such providers to fill a batch and no healthy +// provider is ever refreshed again, while this endpoint goes on returning a +// full, plausible-looking batch. The in-memory ttl cache this replaced was +// incidentally immune, because it marked a provider probed whether or not the +// probe worked; moving the schedule server-side dropped that protection, and +// provider_egress_probe_attempt is what restores it. +// +// Both cutoffs are computed by the caller in Go and bound as parameters: +// observed_at and attempt_at are naive `timestamp` columns holding utc, and +// comparing them against sql now() would cast through the session timezone and +// silently skip a window. +// +// # Two passes, not one +// +// Expressed as a single statement this is a scan of +// network_client_location_reliability with two LEFT JOINs, sorted on +// observed_at from an outer-joined table. That sort cannot use an index: the +// column being ordered on does not exist for most of the rows being ordered. +// At beta's 40 providers that is free. At 100k it is a full scan plus an +// unindexable sort, on every poll. +// +// The ordering makes the split possible. `NULLS FIRST` means every never-probed +// provider sorts ahead of every probed one, so the result is always the +// concatenation of two independently ordered groups: +// +// 1. never probed -- no provider_egress_location row at all. This is the +// dominant group (it is why the ordering is NULLS FIRST), and within it +// every observed_at is equally absent, so the order is client_id alone. As +// an anti-join with no outer-joined column in the ORDER BY it is an ordered +// index scan over (valid, connected, client_id) with a LIMIT: no sort, and +// it stops as soon as the batch is full. +// 2. stale but probed -- has a row, older than minObservedAt. Only reached +// when pass 1 came up short of the limit. Driven from +// provider_egress_location itself, where observed_at is a real, indexable +// column: an ordered range scan over (observed_at, client_id). +// +// Both passes carry the same eligibility predicates, so the concatenation is +// row-for-row what the single statement returned, in the same order, under the +// same limit. `attempt_at IS NULL OR attempt_at < $n` becomes the equivalent +// `NOT EXISTS (... AND $n <= attempt_at)` -- equivalent because client_id is the +// primary key of provider_egress_probe_attempt, so there is at most one row to +// quantify over. The same holds for `observed_at IS NULL` on +// provider_egress_location, whose client_id is likewise a primary key and whose +// observed_at is NOT NULL: the only way that test is true is that no row exists. +func GetProviderEgressLocationDue( + ctx context.Context, + minObservedAt time.Time, + minAttemptAt time.Time, + limit int, +) []server.Id { + clientIds := []server.Id{} + server.Db(ctx, func(conn server.PgConn) { + // pass 1: never probed. Ordered by client_id alone -- every row in this + // group has no observed_at, so the ORDER BY's leading key is constant + // across it and the tie-break is the whole ordering. + // + // `limit` is passed through as given rather than clamped, so a + // nonsensical limit fails exactly as the single-statement version did + // (LIMIT 0 returns nothing; a negative limit is an error). + result, err := conn.Query( + ctx, + ` + SELECT + network_client_location_reliability.client_id + FROM network_client_location_reliability + + WHERE + network_client_location_reliability.connected = true AND + network_client_location_reliability.valid = true AND + EXISTS ( + SELECT 1 FROM provide_key + WHERE + provide_key.client_id = network_client_location_reliability.client_id AND + provide_key.provide_mode = $1 + ) AND + NOT EXISTS ( + SELECT 1 FROM provider_egress_location + WHERE + provider_egress_location.client_id = network_client_location_reliability.client_id + ) AND + NOT EXISTS ( + SELECT 1 FROM provider_egress_probe_attempt + WHERE + provider_egress_probe_attempt.client_id = network_client_location_reliability.client_id AND + $2 <= provider_egress_probe_attempt.attempt_at + ) + + ORDER BY network_client_location_reliability.client_id ASC + LIMIT $3 + `, + ProvideModePublic, + minAttemptAt.UTC(), + limit, + ) + server.WithPgResult(result, err, func() { + for result.Next() { + var clientId server.Id + server.Raise(result.Scan(&clientId)) + clientIds = append(clientIds, clientId) + } + }) + + remaining := limit - len(clientIds) + if remaining <= 0 { + // the batch is full from never-probed providers alone, which is the + // steady state until the population has been swept once. The + // single-statement version would have returned exactly these rows + // too: they all sort ahead of anything with an observed_at. + return + } + + // pass 2: stale but probed. Driven from provider_egress_location, so + // observed_at is a real column of the driving table and the ORDER BY is + // an ordered index scan rather than a sort. + result, err = conn.Query( + ctx, + ` + SELECT + provider_egress_location.client_id + FROM provider_egress_location + + INNER JOIN network_client_location_reliability ON + network_client_location_reliability.client_id = provider_egress_location.client_id + + WHERE + provider_egress_location.observed_at < $2 AND + network_client_location_reliability.connected = true AND + network_client_location_reliability.valid = true AND + EXISTS ( + SELECT 1 FROM provide_key + WHERE + provide_key.client_id = provider_egress_location.client_id AND + provide_key.provide_mode = $1 + ) AND + NOT EXISTS ( + SELECT 1 FROM provider_egress_probe_attempt + WHERE + provider_egress_probe_attempt.client_id = provider_egress_location.client_id AND + $3 <= provider_egress_probe_attempt.attempt_at + ) + + -- oldest probe first, client_id breaking the tie, so batch + -- composition is deterministic instead of plan-dependent + ORDER BY + provider_egress_location.observed_at ASC, + provider_egress_location.client_id ASC + LIMIT $4 + `, + ProvideModePublic, + minObservedAt.UTC(), + minAttemptAt.UTC(), + remaining, + ) + server.WithPgResult(result, err, func() { + // the two passes are separate statements and so separate snapshots. + // A provider that gains its first provider_egress_location row + // between them would be never-probed to pass 1 and stale to pass 2; + // the single-statement version could not do that, so screen it out + // rather than hand the prober the same client twice. + seen := map[server.Id]bool{} + for _, clientId := range clientIds { + seen[clientId] = true + } + for result.Next() { + var clientId server.Id + server.Raise(result.Scan(&clientId)) + if seen[clientId] { + continue + } + clientIds = append(clientIds, clientId) + } + }) + }) + return clientIds +} + // RemoveExpiredProviderEgressLocations drops entries probed before // minObservedAt. func RemoveExpiredProviderEgressLocations(ctx context.Context, minObservedAt time.Time) { @@ -285,3 +830,17 @@ func RemoveExpiredProviderEgressLocations(ctx context.Context, minObservedAt tim )) }) } + +// RemoveExpiredProviderEgressProbeAttempts drops attempts older than +// minAttemptAt. An attempt only carries information for as long as it defers +// the provider (ProviderEgressProbeAttemptBackoff); past that the row is just +// storage held for a client id that may no longer exist. +func RemoveExpiredProviderEgressProbeAttempts(ctx context.Context, minAttemptAt time.Time) { + server.MaintenanceTx(ctx, func(tx server.PgTx) { + server.RaisePgResult(tx.Exec( + ctx, + `DELETE FROM provider_egress_probe_attempt WHERE attempt_at < $1`, + minAttemptAt.UTC(), + )) + }) +} diff --git a/model/provider_egress_location_model_test.go b/model/provider_egress_location_model_test.go index 874f5391..35c789c7 100644 --- a/model/provider_egress_location_model_test.go +++ b/model/provider_egress_location_model_test.go @@ -2,6 +2,8 @@ package model import ( "context" + "fmt" + "slices" "testing" "time" @@ -213,3 +215,453 @@ func TestRemoveExpiredProviderEgressLocations(t *testing.T) { } }) } + +// testing_connectProbeableProvider stands up the minimum a client needs to +// look like a live provider to the due-selection query: a device, a live +// connection with a resolved location, and a provide key of the given mode. +// The caller must run UpdateClientLocationReliabilities afterward -- that is +// what rolls the live connection tables up into the +// network_client_location_reliability row (connected + valid) the query reads. +// It returns the connection id, so a caller can disconnect the provider again. +func testing_connectProbeableProvider( + t testing.TB, + ctx context.Context, + clientId server.Id, + locationId server.Id, + clientAddress string, + provideMode ProvideMode, +) server.Id { + Testing_CreateDevice(ctx, server.NewId(), server.NewId(), clientId, "", "") + + handlerId := CreateNetworkClientHandler(ctx) + connectionId, _, _, _, err := ConnectNetworkClient(ctx, clientId, clientAddress, handlerId) + if err != nil { + t.Fatalf("connect client: %s", err) + } + + if err := SetConnectionLocation(ctx, connectionId, locationId, &ConnectionLocationScores{}); err != nil { + t.Fatalf("set connection location: %s", err) + } + + SetProvide(ctx, clientId, map[ProvideMode][]byte{ + provideMode: []byte("provide-secret"), + }) + + return connectionId +} + +// The prober asks the server what to probe next. The answer must be sourced +// from the live provider population and not from provider_egress_location, +// because the dominant case -- a provider that has never been probed at all -- +// has no row there. +func TestGetProviderEgressLocationDue(t *testing.T) { + server.DefaultTestEnv().Run(t, func(t testing.TB) { + ctx := context.Background() + now := server.NowUtc() + + city := &Location{ + LocationType: LocationTypeCity, + City: "Palo Alto", + Region: "California", + Country: "United States", + CountryCode: "us", + } + CreateLocation(ctx, city) + + fresh := server.NewId() + stale := server.NewId() + never := server.NewId() + // a provider that cannot serve a stranger is unprobeable: the tunnel + // contract would be refused, so it must never be offered to the prober + nonPublic := server.NewId() + + testing_connectProbeableProvider(t, ctx, fresh, city.LocationId, "0.0.0.1:0", ProvideModePublic) + testing_connectProbeableProvider(t, ctx, stale, city.LocationId, "0.0.0.2:0", ProvideModePublic) + testing_connectProbeableProvider(t, ctx, never, city.LocationId, "0.0.0.3:0", ProvideModePublic) + testing_connectProbeableProvider(t, ctx, nonPublic, city.LocationId, "0.0.0.4:0", ProvideModeNetwork) + + UpdateClientLocationReliabilities(ctx, now.Add(-time.Hour), now) + + SetProviderEgressLocation(ctx, &ProviderEgressLocation{ + ClientId: fresh, LocationId: city.LocationId, + CountryCode: "us", ObservedAt: now.Add(-1 * time.Hour), + }) + SetProviderEgressLocation(ctx, &ProviderEgressLocation{ + ClientId: stale, LocationId: city.LocationId, + CountryCode: "us", ObservedAt: now.Add(-72 * time.Hour), + }) + // `never` and `nonPublic` deliberately get no row at all + + // no attempt rows exist in this test, so the attempt cutoff never + // excludes anything; freshness is the only variable + due := GetProviderEgressLocationDue(ctx, now.Add(-24*time.Hour), now, 100) + + // a provider probed an hour ago must not be re-probed; one probed three + // days ago must be; one never probed must be + if slices.Contains(due, fresh) { + t.Fatalf("due = %v, must not contain the provider probed an hour ago (%s)", due, fresh) + } + if !slices.Contains(due, stale) { + t.Fatalf("due = %v, must contain the provider probed three days ago (%s)", due, stale) + } + if !slices.Contains(due, never) { + t.Fatalf("due = %v, must contain the never-probed provider (%s)", due, never) + } + // unprobeable regardless of freshness + if slices.Contains(due, nonPublic) { + t.Fatalf("due = %v, must not contain the provider without a Public provide key (%s)", due, nonPublic) + } + + // oldest first, so the longest-unprobed are probed first: the + // never-probed provider sorts ahead of the three-days-stale one + neverIndex := slices.Index(due, never) + staleIndex := slices.Index(due, stale) + if staleIndex < neverIndex { + t.Fatalf("never-probed provider at %d must sort before the stale one at %d", neverIndex, staleIndex) + } + + // limit is honoured + limited := GetProviderEgressLocationDue(ctx, now.Add(-24*time.Hour), now, 1) + if len(limited) != 1 { + t.Fatalf("len(due) = %d for limit 1, want 1", len(limited)) + } + if limited[0] != never { + t.Fatalf("due[0] = %s for limit 1, want the never-probed provider %s", limited[0], never) + } + }) +} + +// A provider that connects, holds a Public provide key and fails every probe +// never gets a provider_egress_location row, so its observed_at stays NULL, so +// it sorts ahead of every stale-but-refreshable provider -- forever, on every +// poll. Enough of them to fill a batch and no healthy provider's location is +// ever refreshed again, silently: the endpoint keeps returning a full, +// plausible-looking batch of the same dead providers. +// +// A recent attempt must therefore defer a provider exactly as a fresh success +// does. Deleting the attempt predicate from GetProviderEgressLocationDue must +// fail this test. +func TestGetProviderEgressLocationDueDefersRecentlyAttempted(t *testing.T) { + server.DefaultTestEnv().Run(t, func(t testing.TB) { + ctx := context.Background() + now := server.NowUtc() + + city := &Location{ + LocationType: LocationTypeCity, + City: "Palo Alto", + Region: "California", + Country: "United States", + CountryCode: "us", + } + CreateLocation(ctx, city) + + // never probed successfully, and the prober just tried it and failed + dead := server.NewId() + // probed successfully three days ago, never attempted since: the + // provider that actually needs the next probe slot + healthyStale := server.NewId() + + testing_connectProbeableProvider(t, ctx, dead, city.LocationId, "0.0.0.1:0", ProvideModePublic) + testing_connectProbeableProvider(t, ctx, healthyStale, city.LocationId, "0.0.0.2:0", ProvideModePublic) + + UpdateClientLocationReliabilities(ctx, now.Add(-time.Hour), now) + + SetProviderEgressLocation(ctx, &ProviderEgressLocation{ + ClientId: healthyStale, LocationId: city.LocationId, + CountryCode: "us", ObservedAt: now.Add(-72 * time.Hour), + }) + // `dead` deliberately gets no location row -- it has never succeeded -- + // only a failed attempt seconds ago + SetProviderEgressProbeAttempt(ctx, &ProviderEgressProbeAttempt{ + ClientId: dead, + AttemptAt: now.Add(-5 * time.Second), + ProbeFailure: "tunnel_failed", + }) + + minObservedAt := now.Add(-24 * time.Hour) + minAttemptAt := now.Add(-ProviderEgressProbeAttemptBackoff) + + due := GetProviderEgressLocationDue(ctx, minObservedAt, minAttemptAt, 100) + + if slices.Contains(due, dead) { + t.Fatalf("due = %v, must not contain the provider attempted seconds ago (%s)", due, dead) + } + if !slices.Contains(due, healthyStale) { + t.Fatalf("due = %v, must contain the stale-but-refreshable provider (%s)", due, healthyStale) + } + + // the starvation itself: with a batch big enough for exactly one + // provider, the slot must go to the one that can actually be refreshed, + // not to the never-probed one that just failed. Without the attempt + // predicate `dead` wins this on observed_at IS NULL every single poll. + limited := GetProviderEgressLocationDue(ctx, minObservedAt, minAttemptAt, 1) + if len(limited) != 1 { + t.Fatalf("len(due) = %d for limit 1, want 1", len(limited)) + } + if limited[0] != healthyStale { + t.Fatalf("due[0] = %s for limit 1, want the refreshable provider %s, not the just-failed one %s", limited[0], healthyStale, dead) + } + + // ... and the deferral is a backoff, not a ban: once the backoff has + // elapsed the same provider is offered again. The caller computes the + // cutoff as (wall clock - backoff), so a poll exactly one backoff period + // after the attempt computes `now`. + afterBackoff := GetProviderEgressLocationDue(ctx, minObservedAt, now, 100) + if !slices.Contains(afterBackoff, dead) { + t.Fatalf("due = %v, must contain the failed provider (%s) again once the attempt backoff has elapsed", afterBackoff, dead) + } + }) +} + +// Only live, routable providers are probeable. A provider that has gone offline +// (connected = false) or that looks messed up from a routing perspective +// (valid = false, a generated column: more than one address hash or location on +// its live connections) must not be handed to the prober. Deleting either +// predicate from GetProviderEgressLocationDue must fail this test. +func TestGetProviderEgressLocationDueRequiresConnectedAndValid(t *testing.T) { + server.DefaultTestEnv().Run(t, func(t testing.TB) { + ctx := context.Background() + now := server.NowUtc() + + city := &Location{ + LocationType: LocationTypeCity, + City: "Palo Alto", + Region: "California", + Country: "United States", + CountryCode: "us", + } + CreateLocation(ctx, city) + + good := server.NewId() + disconnected := server.NewId() + invalid := server.NewId() + + testing_connectProbeableProvider(t, ctx, good, city.LocationId, "0.0.0.1:0", ProvideModePublic) + disconnectedConnectionId := testing_connectProbeableProvider(t, ctx, disconnected, city.LocationId, "0.0.0.2:0", ProvideModePublic) + + // `invalid` holds two simultaneous connections from two different + // addresses, which makes client_address_hash_count = 2 and so the + // generated `valid` column false. The two addresses must be in + // different /29s: server.ClientIpHash buckets ipv4 to the /29 network, + // so e.g. 0.0.0.3 and 0.0.0.4 would hash the same and count as one. + testing_connectProbeableProvider(t, ctx, invalid, city.LocationId, "0.0.0.3:0", ProvideModePublic) + secondHandlerId := CreateNetworkClientHandler(ctx) + secondConnectionId, _, _, _, err := ConnectNetworkClient(ctx, invalid, "0.0.8.3:0", secondHandlerId) + if err != nil { + t.Fatalf("connect second address: %s", err) + } + if err := SetConnectionLocation(ctx, secondConnectionId, city.LocationId, &ConnectionLocationScores{}); err != nil { + t.Fatalf("set second connection location: %s", err) + } + + // first roll-up: everything above is connected + UpdateClientLocationReliabilities(ctx, now.Add(-time.Hour), now) + + // `disconnected` drops off, and a second roll-up flips its reliability + // row's connected to false (the row itself survives) + if err := DisconnectNetworkClient(ctx, disconnectedConnectionId); err != nil { + t.Fatalf("disconnect client: %s", err) + } + UpdateClientLocationReliabilities(ctx, now.Add(-time.Hour), server.NowUtc()) + + due := GetProviderEgressLocationDue(ctx, now.Add(-24*time.Hour), now, 100) + + if !slices.Contains(due, good) { + t.Fatalf("due = %v, must contain the connected, valid provider (%s)", due, good) + } + if slices.Contains(due, disconnected) { + t.Fatalf("due = %v, must not contain the disconnected provider (%s)", due, disconnected) + } + if slices.Contains(due, invalid) { + t.Fatalf("due = %v, must not contain the provider whose reliability row is not valid (%s)", due, invalid) + } + }) +} + +// The attempt upsert is monotonic in attempt_at, for the same reason the +// location upsert is: a replayed or out-of-order report must not move the last +// attempt backwards and hand the provider back to the prober early. +func TestProviderEgressProbeAttemptUpsertIgnoresOlderReplay(t *testing.T) { + server.DefaultTestEnv().Run(t, func(t testing.TB) { + ctx := context.Background() + + clientId := server.NewId() + newer := server.NowUtc() + older := newer.Add(-time.Hour) + + SetProviderEgressProbeAttempt(ctx, &ProviderEgressProbeAttempt{ + ClientId: clientId, AttemptAt: newer, ProbeFailure: "no_consensus", + }) + SetProviderEgressProbeAttempt(ctx, &ProviderEgressProbeAttempt{ + ClientId: clientId, AttemptAt: older, ProbeFailure: "tunnel_failed", + }) + + got := GetProviderEgressProbeAttempt(ctx, clientId) + if got == nil { + t.Fatal("expected a stored probe attempt") + } + assert.Equal(t, got.ProbeFailure, "no_consensus") + // postgres `timestamp` keeps microseconds, Go keeps nanoseconds, so + // compare with a tolerance rather than for equality + if delta := got.AttemptAt.Sub(newer); delta < -time.Millisecond || time.Millisecond < delta { + t.Fatalf("attempt_at = %s, want the newer attempt %s", got.AttemptAt, newer) + } + + // a strictly newer report does win + newest := newer.Add(time.Minute) + SetProviderEgressProbeAttempt(ctx, &ProviderEgressProbeAttempt{ + ClientId: clientId, AttemptAt: newest, ProbeFailure: "", + }) + got = GetProviderEgressProbeAttempt(ctx, clientId) + assert.Equal(t, got.ProbeFailure, "") + + // absent + if GetProviderEgressProbeAttempt(ctx, server.NewId()) != nil { + t.Fatal("absent attempt must return nil") + } + }) +} + +func TestRemoveExpiredProviderEgressProbeAttempts(t *testing.T) { + server.DefaultTestEnv().Run(t, func(t testing.TB) { + ctx := context.Background() + + keep := server.NewId() + drop := server.NewId() + SetProviderEgressProbeAttempt(ctx, &ProviderEgressProbeAttempt{ + ClientId: keep, AttemptAt: server.NowUtc(), + }) + SetProviderEgressProbeAttempt(ctx, &ProviderEgressProbeAttempt{ + ClientId: drop, AttemptAt: server.NowUtc().Add(-30 * 24 * time.Hour), + }) + + RemoveExpiredProviderEgressProbeAttempts(ctx, server.NowUtc().Add(-24*time.Hour)) + + if GetProviderEgressProbeAttempt(ctx, keep) == nil { + t.Fatal("recent attempt must survive the sweep") + } + if GetProviderEgressProbeAttempt(ctx, drop) != nil { + t.Fatal("old attempt must be swept") + } + }) +} + +// GetProviderEgressLocationDue is served by two statements -- never-probed +// first, then stale-but-probed only when the first came up short -- because the +// single-statement form sorts on observed_at from an outer-joined table, which +// cannot use an index and becomes a full scan plus an unindexable sort at 100k +// providers. +// +// The split is only safe if the concatenation is row-for-row what one statement +// returned, at every limit. That is what this asserts: it builds one population +// covering every eligibility case and then walks the limit from 0 past the end, +// requiring each result to be exactly the prefix of the full ordering. Limit 3 +// is the seam (pass one exactly fills the batch) and limit 4 is the first that +// crosses into pass two. +func TestGetProviderEgressLocationDueOrderingIsStableAcrossLimits(t *testing.T) { + server.DefaultTestEnv().Run(t, func(t testing.TB) { + ctx := context.Background() + now := server.NowUtc() + + city := &Location{ + LocationType: LocationTypeCity, + City: "Palo Alto", + Region: "California", + Country: "United States", + CountryCode: "us", + } + CreateLocation(ctx, city) + + // three never-probed providers: the dominant group, and the reason the + // ordering is NULLS FIRST + never := []server.Id{server.NewId(), server.NewId(), server.NewId()} + // two stale ones, at different ages -- the older must be handed out first + staleOlder := server.NewId() + staleNewer := server.NewId() + // probed an hour ago: not due + fresh := server.NewId() + // never probed, but attempted seconds ago: deferred by the backoff, and + // the case that would otherwise starve the queue + attempted := server.NewId() + // probed long ago AND attempted seconds ago. This one is only screened + // by the backoff predicate on the stale-but-probed pass -- the + // never-probed pass never sees it, because it has an egress row. Drop + // that predicate and it reappears in the batch. + staleAttempted := server.NewId() + // no Public provide key: unprobeable at any freshness + nonPublic := server.NewId() + + address := 0 + connectProvider := func(clientId server.Id, provideMode ProvideMode) { + address += 1 + testing_connectProbeableProvider( + t, ctx, clientId, city.LocationId, + fmt.Sprintf("0.0.%d.1:0", address), provideMode, + ) + } + for _, clientId := range never { + connectProvider(clientId, ProvideModePublic) + } + connectProvider(staleOlder, ProvideModePublic) + connectProvider(staleNewer, ProvideModePublic) + connectProvider(fresh, ProvideModePublic) + connectProvider(attempted, ProvideModePublic) + connectProvider(staleAttempted, ProvideModePublic) + connectProvider(nonPublic, ProvideModeNetwork) + + UpdateClientLocationReliabilities(ctx, now.Add(-time.Hour), now) + + SetProviderEgressLocation(ctx, &ProviderEgressLocation{ + ClientId: staleOlder, LocationId: city.LocationId, + CountryCode: "us", ObservedAt: now.Add(-100 * time.Hour), + }) + SetProviderEgressLocation(ctx, &ProviderEgressLocation{ + ClientId: staleNewer, LocationId: city.LocationId, + CountryCode: "us", ObservedAt: now.Add(-50 * time.Hour), + }) + SetProviderEgressLocation(ctx, &ProviderEgressLocation{ + ClientId: fresh, LocationId: city.LocationId, + CountryCode: "us", ObservedAt: now.Add(-1 * time.Hour), + }) + SetProviderEgressLocation(ctx, &ProviderEgressLocation{ + ClientId: staleAttempted, LocationId: city.LocationId, + CountryCode: "us", ObservedAt: now.Add(-200 * time.Hour), + }) + SetProviderEgressProbeAttempt(ctx, &ProviderEgressProbeAttempt{ + ClientId: attempted, AttemptAt: now.Add(-5 * time.Second), + ProbeFailure: "tunnel_failed", + }) + // oldest observed_at of all, so it would sort to the head of the + // stale group if the backoff did not exclude it + SetProviderEgressProbeAttempt(ctx, &ProviderEgressProbeAttempt{ + ClientId: staleAttempted, AttemptAt: now.Add(-5 * time.Second), + ProbeFailure: "tunnel_failed", + }) + + minObservedAt := now.Add(-24 * time.Hour) + minAttemptAt := now.Add(-ProviderEgressProbeAttemptBackoff) + + // the never-probed group ties on a missing observed_at, so client_id + // alone orders it -- and postgres orders uuid by bytes, which is what + // server.Id.Cmp does + expected := slices.Clone(never) + slices.SortFunc(expected, func(a server.Id, b server.Id) int { return a.Cmp(b) }) + // ... then the probed group, oldest probe first + expected = append(expected, staleOlder, staleNewer) + + due := GetProviderEgressLocationDue(ctx, minObservedAt, minAttemptAt, 100) + if !slices.Equal(due, expected) { + t.Fatalf("due = %v, want %v (never-probed by client_id, then stale oldest-first; fresh/attempted/stale-attempted/non-public excluded)", due, expected) + } + + // every limit must return exactly the prefix of that ordering. limit 3 + // is the pass-one/pass-two seam; 4 is the first to cross it. + for limit := 0; limit <= len(expected)+2; limit += 1 { + want := expected[:min(limit, len(expected))] + got := GetProviderEgressLocationDue(ctx, minObservedAt, minAttemptAt, limit) + if !slices.Equal(got, want) { + t.Errorf("limit %d: due = %v, want %v", limit, got, want) + } + } + }) +} diff --git a/taskworker/work/provider_egress_location_work.go b/taskworker/work/provider_egress_location_work.go index 94901c34..533d9609 100644 --- a/taskworker/work/provider_egress_location_work.go +++ b/taskworker/work/provider_egress_location_work.go @@ -33,8 +33,14 @@ func RemoveExpiredProviderEgressLocations( _ *RemoveExpiredProviderEgressLocationsArgs, clientSession *session.ClientSession, ) (*RemoveExpiredProviderEgressLocationsResult, error) { - minObservedAt := server.NowUtc().Add(-4 * model.ProviderEgressLocationMaxAge) + now := server.NowUtc() + minObservedAt := now.Add(-4 * model.ProviderEgressLocationMaxAge) model.RemoveExpiredProviderEgressLocations(clientSession.Ctx, minObservedAt) + // probe attempts stop meaning anything once they no longer defer the + // provider; same reasoning as above, a looser multiple of the window that + // actually matters. + minAttemptAt := now.Add(-4 * model.ProviderEgressProbeAttemptBackoff) + model.RemoveExpiredProviderEgressProbeAttempts(clientSession.Ctx, minAttemptAt) return &RemoveExpiredProviderEgressLocationsResult{}, nil }