From a80921c354bc13db479fcfd4f24fb7046eb877aa Mon Sep 17 00:00:00 2001 From: Ryan Mello Date: Sun, 26 Jul 2026 09:09:23 +0100 Subject: [PATCH 1/4] feat(api): serve the provider egress probe schedule from the server The operator's prober decided what to probe from an in-memory ttl cache, so a restart re-probed the whole provider population and nothing durable recorded what was actually due. The freshness data already exists server-side in provider_egress_location.observed_at; this exposes it so the server owns the schedule. model.GetProviderEgressLocationDue sources candidates from the live provider population (network_client_location_reliability, connected + valid) and LEFT JOINs the egress row, rather than selecting from provider_egress_location. The dominant case is a provider that has never been probed and so has no egress row at all; selecting from that table would return exactly the providers that least need probing and none of the ones that most do. Only providers holding a Public provide key are returned. Probing tunnels through the provider itself, i.e. opens a contract from outside the provider's own network, which a provider without a Public key refuses -- offering one to the prober would burn a probe slot on a guaranteed failure. Same EXISTS filter UpdateClientLocations and UpdateClientScores already apply. The staleness cutoff is computed in Go and bound as a query argument. observed_at is a naive `timestamp` holding utc, so comparing it against sql now() would cast through the session timezone and silently skip a window. GET /network/provider-egress-due is operator-to-server, authenticated by the same X-UR-Operator-Secret shared secret as the ingest endpoint (hmac.Equal against the memoized, fail-closed operatorIngestSecret) rather than a network jwt. Its cutoff is half ProviderEgressLocationMaxAge: at the full max age every location would lapse to the mmdb fallback at the exact moment it became due and stay lapsed until the prober reached it. Co-Authored-By: Claude Claude-Session: https://claude.ai/code/session_01QtgqtCmKJRXdsQ5ktiqwkg --- api/api.go | 1 + .../provider_egress_location_handlers.go | 75 +++++++ .../provider_egress_location_handlers_test.go | 192 ++++++++++++++++++ model/provider_egress_location_model.go | 79 +++++++ model/provider_egress_location_model_test.go | 111 ++++++++++ 5 files changed, 458 insertions(+) diff --git a/api/api.go b/api/api.go index cb487962..e284d7af 100644 --- a/api/api.go +++ b/api/api.go @@ -55,6 +55,7 @@ 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("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..7f6feac8 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,76 @@ func ProviderEgressLocationSubmit(w http.ResponseWriter, r *http.Request) { 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. +// +// 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) + } + + // the cutoff is computed here and passed as an argument; observed_at is a + // naive timestamp holding utc, so comparing it to sql now() in the query + // would cast through the session timezone + minObservedAt := server.NowUtc().Add(-providerEgressDueAge) + + result := &ProviderEgressLocationDueResult{ + ClientIds: model.GetProviderEgressLocationDue(r.Context(), minObservedAt, 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..ab298a94 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,191 @@ 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()) + } + }) +} + +// 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/model/provider_egress_location_model.go b/model/provider_egress_location_model.go index 240bb8b4..7bf9ae00 100644 --- a/model/provider_egress_location_model.go +++ b/model/provider_egress_location_model.go @@ -274,6 +274,85 @@ func GetLocation(ctx context.Context, locationId server.Id) *Location { return loc } +// GetProviderEgressLocationDue returns the client ids of providers whose +// egress location is due for a probe: their newest probe is older than +// minObservedAt, or they have never been probed at all. 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. +// +// Two 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). +// +// The cutoff is computed by the caller in Go and bound as a parameter: +// observed_at is a naive `timestamp` holding utc, and comparing it against sql +// now() would cast through the session timezone and silently skip a window. +func GetProviderEgressLocationDue( + ctx context.Context, + minObservedAt time.Time, + limit int, +) []server.Id { + clientIds := []server.Id{} + server.Db(ctx, func(conn server.PgConn) { + result, err := conn.Query( + ctx, + ` + SELECT + network_client_location_reliability.client_id + FROM network_client_location_reliability + + LEFT JOIN provider_egress_location ON + provider_egress_location.client_id = network_client_location_reliability.client_id + + 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 + ( + provider_egress_location.observed_at IS NULL OR + provider_egress_location.observed_at < $2 + ) + + -- never-probed sorts ahead of merely stale: a missing observed_at + -- is infinitely old + ORDER BY provider_egress_location.observed_at ASC NULLS FIRST + LIMIT $3 + `, + ProvideModePublic, + minObservedAt.UTC(), + limit, + ) + server.WithPgResult(result, err, func() { + for result.Next() { + var clientId server.Id + server.Raise(result.Scan(&clientId)) + clientIds = append(clientIds, clientId) + } + }) + }) + return clientIds +} + // RemoveExpiredProviderEgressLocations drops entries probed before // minObservedAt. func RemoveExpiredProviderEgressLocations(ctx context.Context, minObservedAt time.Time) { diff --git a/model/provider_egress_location_model_test.go b/model/provider_egress_location_model_test.go index 874f5391..e883387f 100644 --- a/model/provider_egress_location_model_test.go +++ b/model/provider_egress_location_model_test.go @@ -2,6 +2,7 @@ package model import ( "context" + "slices" "testing" "time" @@ -213,3 +214,113 @@ 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. +func testing_connectProbeableProvider( + t testing.TB, + ctx context.Context, + clientId server.Id, + locationId server.Id, + clientAddress string, + provideMode ProvideMode, +) { + 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"), + }) +} + +// 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 + + due := GetProviderEgressLocationDue(ctx, now.Add(-24*time.Hour), 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), 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) + } + }) +} From b1e2732ccf672809e5145f4ead437f16613fed7d Mon Sep 17 00:00:00 2001 From: Ryan Mello Date: Sun, 26 Jul 2026 09:46:43 +0100 Subject: [PATCH 2/4] fix(model,api): defer providers that fail to probe, not only ones that succeed The only thing that moved a provider off the head of the due queue was a successful probe writing provider_egress_location. A provider that connects and holds a Public provide key but always fails to probe -- firewalled egress, dead upstream, anything other than the missing Public key the query already screens for -- never gets a row, so its observed_at stays NULL, so it sorts ahead of every stale-but-refreshable provider on every poll, forever. Five hundred such providers and a prober asking for limit=500 gets the same five hundred dead providers every time; no healthy provider's location is ever refreshed. It fails silently: the endpoint keeps 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. Record attempts, not just successes. provider_egress_probe_attempt is a small table keyed by client_id -- the attempt cannot live on provider_egress_location, because the case it exists to handle is precisely a provider with no row there. GetProviderEgressLocationDue now requires both no fresh success and no recent attempt, with a much shorter backoff for attempts (6h) than for success freshness (half ProviderEgressLocationMaxAge, 3.5d): a failing provider should be retried periodically, just not on every poll. Both cutoffs are computed in Go and bound as arguments -- attempt_at, like observed_at, is a naive `timestamp` holding utc, and comparing it to sql now() casts through the session timezone. The prober reports an attempt to POST /network/provider-egress-attempt, on the same operator-authenticated surface as the other two endpoints (X-UR-Operator-Secret, hmac.Equal, the memoized fail-closed operatorIngestSecret). The server timestamps the attempt rather than trusting the prober's clock. Attempt rows are swept by the existing expiry task. This pulls probe_attempt_at/probe_failure forward from the P2 data model in docs/superpowers/specs/2026-07-25-enforced-provider-geo-probing-design.md, because P1's schedule cannot function without them. Only those two columns, not the rest of the verdict model. Also adds a client_id secondary sort key, so batch composition under a limit is deterministic rather than plan-dependent -- the whole never-probed population otherwise ties on NULL observed_at. Tests: a provider never probed successfully but attempted seconds ago must not be due, must not take the single batch slot from a stale-but-refreshable provider, and must become due again once the backoff elapses; the same over http via the attempt endpoint. Plus the two coverage gaps review flagged: deleting either the connected or the valid predicate now fails a test, and the handler's staleness cutoff is exercised by a probed (not merely never-probed) provider. Co-Authored-By: Claude Claude-Session: https://claude.ai/code/session_01QtgqtCmKJRXdsQ5ktiqwkg --- api/api.go | 1 + .../provider_egress_location_handlers.go | 71 +++++- .../provider_egress_location_handlers_test.go | 210 ++++++++++++++++ .../provider_egress_location_controller.go | 52 ++++ db_migrations.go | 36 +++ model/provider_egress_location_model.go | 152 +++++++++++- model/provider_egress_location_model_test.go | 226 +++++++++++++++++- .../work/provider_egress_location_work.go | 8 +- 8 files changed, 738 insertions(+), 18 deletions(-) diff --git a/api/api.go b/api/api.go index e284d7af..1ad876ed 100644 --- a/api/api.go +++ b/api/api.go @@ -56,6 +56,7 @@ func Routes() []*router.Route { 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 7f6feac8..264da746 100644 --- a/api/handlers/provider_egress_location_handlers.go +++ b/api/handlers/provider_egress_location_handlers.go @@ -94,6 +94,59 @@ func ProviderEgressLocationSubmit(w http.ResponseWriter, r *http.Request) { } } +// 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. @@ -129,6 +182,12 @@ type ProviderEgressLocationDueResult struct { // 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. @@ -152,13 +211,15 @@ func ProviderEgressLocationDue(w http.ResponseWriter, r *http.Request) { limit = min(parsed, maxProviderEgressDueLimit) } - // the cutoff is computed here and passed as an argument; observed_at is a - // naive timestamp holding utc, so comparing it to sql now() in the query - // would cast through the session timezone - minObservedAt := server.NowUtc().Add(-providerEgressDueAge) + // 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, limit), + ClientIds: model.GetProviderEgressLocationDue(r.Context(), minObservedAt, minAttemptAt, limit), } w.Header().Set("Content-Type", "application/json") diff --git a/api/handlers/provider_egress_location_handlers_test.go b/api/handlers/provider_egress_location_handlers_test.go index ab298a94..0f094ca0 100644 --- a/api/handlers/provider_egress_location_handlers_test.go +++ b/api/handlers/provider_egress_location_handlers_test.go @@ -320,6 +320,216 @@ func TestProviderEgressLocationDueHonoursLimit(t *testing.T) { }) } +// 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 diff --git a/controller/provider_egress_location_controller.go b/controller/provider_egress_location_controller.go index 169de94a..4afc02a7 100644 --- a/controller/provider_egress_location_controller.go +++ b/controller/provider_egress_location_controller.go @@ -143,3 +143,55 @@ func SubmitProviderEgressLocation( 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/db_migrations.go b/db_migrations.go index be206cee..dcb2662e 100644 --- a/db_migrations.go +++ b/db_migrations.go @@ -4523,4 +4523,40 @@ 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) + `), } diff --git a/model/provider_egress_location_model.go b/model/provider_egress_location_model.go index 7bf9ae00..55d0aaf3 100644 --- a/model/provider_egress_location_model.go +++ b/model/provider_egress_location_model.go @@ -13,6 +13,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 +96,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 @@ -275,15 +366,16 @@ func GetLocation(ctx context.Context, locationId server.Id) *Location { } // GetProviderEgressLocationDue returns the client ids of providers whose -// egress location is due for a probe: their newest probe is older than -// minObservedAt, or they have never been probed at all. Oldest first, so the +// 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. // -// Two things about the shape of this query matter. +// 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 @@ -299,12 +391,26 @@ func GetLocation(ctx context.Context, locationId server.Id) *Location { // guaranteed failure. This is the same filter UpdateClientLocations and // UpdateClientScores apply (network_client_location_model.go). // -// The cutoff is computed by the caller in Go and bound as a parameter: -// observed_at is a naive `timestamp` holding utc, and comparing it against sql -// now() would cast through the session timezone and silently skip a window. +// 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. func GetProviderEgressLocationDue( ctx context.Context, minObservedAt time.Time, + minAttemptAt time.Time, limit int, ) []server.Id { clientIds := []server.Id{} @@ -319,6 +425,9 @@ func GetProviderEgressLocationDue( LEFT JOIN provider_egress_location ON provider_egress_location.client_id = network_client_location_reliability.client_id + LEFT JOIN provider_egress_probe_attempt ON + provider_egress_probe_attempt.client_id = network_client_location_reliability.client_id + WHERE network_client_location_reliability.connected = true AND network_client_location_reliability.valid = true AND @@ -331,15 +440,26 @@ func GetProviderEgressLocationDue( ( provider_egress_location.observed_at IS NULL OR provider_egress_location.observed_at < $2 + ) AND + ( + provider_egress_probe_attempt.attempt_at IS NULL OR + provider_egress_probe_attempt.attempt_at < $3 ) -- never-probed sorts ahead of merely stale: a missing observed_at - -- is infinitely old - ORDER BY provider_egress_location.observed_at ASC NULLS FIRST - LIMIT $3 + -- is infinitely old. client_id breaks the tie so batch composition + -- is deterministic instead of plan-dependent -- otherwise the whole + -- never-probed population ties on NULL and which slice of it the + -- prober gets back under a limit is whatever order the executor + -- happened to produce. + ORDER BY + provider_egress_location.observed_at ASC NULLS FIRST, + network_client_location_reliability.client_id ASC + LIMIT $4 `, ProvideModePublic, minObservedAt.UTC(), + minAttemptAt.UTC(), limit, ) server.WithPgResult(result, err, func() { @@ -364,3 +484,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 e883387f..c5189a65 100644 --- a/model/provider_egress_location_model_test.go +++ b/model/provider_egress_location_model_test.go @@ -221,6 +221,7 @@ func TestRemoveExpiredProviderEgressLocations(t *testing.T) { // 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, @@ -228,7 +229,7 @@ func testing_connectProbeableProvider( locationId server.Id, clientAddress string, provideMode ProvideMode, -) { +) server.Id { Testing_CreateDevice(ctx, server.NewId(), server.NewId(), clientId, "", "") handlerId := CreateNetworkClientHandler(ctx) @@ -244,6 +245,8 @@ func testing_connectProbeableProvider( 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 @@ -288,7 +291,9 @@ func TestGetProviderEgressLocationDue(t *testing.T) { }) // `never` and `nonPublic` deliberately get no row at all - due := GetProviderEgressLocationDue(ctx, now.Add(-24*time.Hour), 100) + // 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 @@ -315,7 +320,7 @@ func TestGetProviderEgressLocationDue(t *testing.T) { } // limit is honoured - limited := GetProviderEgressLocationDue(ctx, now.Add(-24*time.Hour), 1) + 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)) } @@ -324,3 +329,218 @@ func TestGetProviderEgressLocationDue(t *testing.T) { } }) } + +// 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") + } + }) +} 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 } From 682b29b4dccbb30db46542087e7d1708b8970055 Mon Sep 17 00:00:00 2001 From: Ryan Mello Date: Mon, 27 Jul 2026 03:22:51 +0100 Subject: [PATCH 3/4] fix(controller,model): the egress probe must match existing locations, never create them SubmitProviderEgressLocation validated a probe-supplied city/region only as trimmed, non-empty and <=128 chars, then handed it to model.CreateLocation. CreateLocation dedupes a city on its exact location_name, so an unrecognised spelling did not fail -- it silently inserted a new permanent row into the shared `location` table and added it to the search index. This is not hypothetical. The prober's consensus stores the winning source's original display string, and the three free geolocation APIs demonstrably disagree on spelling: we observed "Frankfurt am Main (Innenstadt I)" against "Frankfurt am Main" for the same host. "Frankfurt am Main", "Frankfurt Am Main" and "Frankfurt/Main" would each become their own row. Those rows are permanent, they survive a code revert, they feed the location search and the provider list, and there is no cleanup path. An ingest endpoint that can add rows to that table is an endpoint that can corrupt it from outside. A geolocation probe has no business defining the world's cities, so the ingest path now resolves against rows that already exist and creates none. model.MatchExistingLocation walks country -> region -> city, trying an exact, fully-indexed match at each level first (the common case: the winning source usually spells it the way the mmdb import did) and falling back to a normalized comparison -- lowercased with punctuation and whitespace dropped -- so the trivial variants land on the row that is already there. Standard library only: a transliteration/fuzzy-match dependency is a lot of new behaviour to take on for a path whose failure mode is already "use the country". Deliberately conservative -- "Frankfurt/Main" folds to "frankfurtmain", does not match "frankfurtammain", and falls back rather than guessing at the wrong row. When nothing resolves, the submission is stored at country granularity instead of being rejected: country is the granularity this design treats as trustworthy, and losing city precision for one probe beats polluting shared data permanently. The country fallback still goes through CreateLocation -- country rows are keyed on country_code, so a variant *name* cannot produce a second row for the same country the way a variant city name can, and a probe from a country not yet in the table must not be dropped. city_confident now records the granularity actually stored rather than what the probe claimed, keeping the schema's documented invariant intact: location_id is a city row exactly when city_confident is set. Tests: an unmatched city stores country granularity and leaves the location row count unchanged; five spelling/case/punctuation variants of one city all resolve to the single existing row and add none. Before this change the unmatched city took the table 3 -> 4 rows and stored as a city, and the variants took it 3 -> 7. Co-Authored-By: Claude Claude-Session: https://claude.ai/code/session_01QtgqtCmKJRXdsQ5ktiqwkg --- .../provider_egress_location_controller.go | 54 +++- ...rovider_egress_location_controller_test.go | 173 ++++++++++- model/provider_egress_location_model.go | 280 ++++++++++++++++++ 3 files changed, 494 insertions(+), 13 deletions(-) diff --git a/controller/provider_egress_location_controller.go b/controller/provider_egress_location_controller.go index 4afc02a7..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,7 +167,7 @@ func SubmitProviderEgressLocation( Hosting: args.Hosting, Proxy: args.Proxy, Mobile: args.Mobile, - CityConfident: args.CityConfident, + CityConfident: cityConfident, ObservedAt: args.ObservedAt, }) 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/model/provider_egress_location_model.go b/model/provider_egress_location_model.go index 55d0aaf3..1bb13eed 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" ) @@ -365,6 +366,252 @@ 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 @@ -407,6 +654,39 @@ func GetLocation(ctx context.Context, locationId server.Id) *Location { // 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, From 37ca6cd7202afbb2327fae5681cdd9342cc9b88e Mon Sep 17 00:00:00 2001 From: Ryan Mello Date: Mon, 27 Jul 2026 03:23:25 +0100 Subject: [PATCH 4/4] perf(model): split the provider-egress due query so the common case avoids a sort GetProviderEgressLocationDue scanned network_client_location_reliability, added two LEFT JOINs and an EXISTS, then 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 it is free. At 100k it is a full scan plus an unindexable sort on every poll. Measured on a 100k-provider dataset (30% probed, 5% recently attempted): before Parallel Seq Scan over all 100k rows, hash-joins both tables, quicksort of ~94k rows (3.3MB), 2282 shared buffers, 62.9 ms after pass 1: Index Only Scan + two Merge Anti Joins, no sort, 138 rows touched, 18 shared buffers, 0.34 ms pass 2: Index Only Scan on the new index, no sort, 350 shared buffers, 1.6 ms The ordering is what 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 -- the dominant group, which is why the ordering is NULLS FIRST. Within it observed_at is equally absent, so client_id alone orders it, and as an anti-join with no outer-joined column in the ORDER BY it is an ordered index scan over the existing (valid, connected, client_id) index that stops as soon as the batch is full. 2. stale but probed -- only queried when pass 1 came up short of the limit. Driven from provider_egress_location, where observed_at is a real, indexable column. `attempt_at IS NULL OR attempt_at < $n` becomes `NOT EXISTS (... AND $n <= attempt_at)`, equivalent because client_id is the primary key of provider_egress_probe_attempt. Same for `observed_at IS NULL`, whose table also keys on client_id and whose observed_at is NOT NULL -- the only way that test is true is that no row exists. The two passes are separate snapshots, so a client that gains its first egress row between them is screened out rather than handed to the prober twice; the single statement could not do that. Verified row-for-row identical to the single statement on the 100k dataset at limits 0, 1, 2, 50, 100, 1000, 20000, 69000, 69500, 70000, 70500 and 100000 -- the middle four straddle the pass-one/pass-two seam (the never-probed population is ~70138) and the last exhausts the eligible set at 94088. New migration adds (observed_at, client_id) on provider_egress_location so pass 2's predicate and its full ORDER BY, tie-break included, are one ordered index scan; without it the planner adds an Incremental Sort (936 buffers vs 350). Appended, never inserted -- migrations apply by slice index here, so editing or reordering an applied one corrupts live databases. Pass 1 needs no new index. TestGetProviderEgressLocationDueOrderingIsStableAcrossLimits asserts the full ordering and then that every limit from 0 past the end returns exactly its prefix, over a population covering never-probed, stale, fresh, recently-attempted, stale-AND-recently-attempted and non-public. It passes against the old single statement as well as the new pair -- it is a behaviour-preservation test, not a description of the new code -- and fails against a pass 2 that uses the full limit instead of the remainder, and against a pass 2 that drops the attempt backoff. The stale-AND-attempted provider exists because of that second check: without it nothing in the suite covered the only case pass 2's backoff predicate screens, and the broken variant passed. Co-Authored-By: Claude Claude-Session: https://claude.ai/code/session_01QtgqtCmKJRXdsQ5ktiqwkg --- db_migrations.go | 27 +++++ model/provider_egress_location_model.go | 108 +++++++++++++---- model/provider_egress_location_model_test.go | 121 +++++++++++++++++++ 3 files changed, 235 insertions(+), 21 deletions(-) diff --git a/db_migrations.go b/db_migrations.go index dcb2662e..c1bbe401 100644 --- a/db_migrations.go +++ b/db_migrations.go @@ -4559,4 +4559,31 @@ var migrations = []any{ 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 1bb13eed..88ee8603 100644 --- a/model/provider_egress_location_model.go +++ b/model/provider_egress_location_model.go @@ -695,6 +695,13 @@ func GetProviderEgressLocationDue( ) []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, ` @@ -702,12 +709,6 @@ func GetProviderEgressLocationDue( network_client_location_reliability.client_id FROM network_client_location_reliability - LEFT JOIN provider_egress_location ON - provider_egress_location.client_id = network_client_location_reliability.client_id - - LEFT JOIN provider_egress_probe_attempt ON - provider_egress_probe_attempt.client_id = network_client_location_reliability.client_id - WHERE network_client_location_reliability.connected = true AND network_client_location_reliability.valid = true AND @@ -717,35 +718,100 @@ func GetProviderEgressLocationDue( provide_key.client_id = network_client_location_reliability.client_id AND provide_key.provide_mode = $1 ) AND - ( - provider_egress_location.observed_at IS NULL OR - provider_egress_location.observed_at < $2 + 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 - ( - provider_egress_probe_attempt.attempt_at IS NULL OR - provider_egress_probe_attempt.attempt_at < $3 + 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 ) - -- never-probed sorts ahead of merely stale: a missing observed_at - -- is infinitely old. client_id breaks the tie so batch composition - -- is deterministic instead of plan-dependent -- otherwise the whole - -- never-probed population ties on NULL and which slice of it the - -- prober gets back under a limit is whatever order the executor - -- happened to produce. + -- 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 NULLS FIRST, - network_client_location_reliability.client_id ASC + provider_egress_location.observed_at ASC, + provider_egress_location.client_id ASC LIMIT $4 `, ProvideModePublic, minObservedAt.UTC(), minAttemptAt.UTC(), - limit, + 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) } }) diff --git a/model/provider_egress_location_model_test.go b/model/provider_egress_location_model_test.go index c5189a65..35c789c7 100644 --- a/model/provider_egress_location_model_test.go +++ b/model/provider_egress_location_model_test.go @@ -2,6 +2,7 @@ package model import ( "context" + "fmt" "slices" "testing" "time" @@ -544,3 +545,123 @@ func TestRemoveExpiredProviderEgressProbeAttempts(t *testing.T) { } }) } + +// 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) + } + } + }) +}