feat(api): serve the provider egress probe schedule from the server

api/api.go

@@ -55,6 +55,8 @@ func Routes() []*router.Route {
 		router.NewRoute("POST", "/network/remove-client", handlers.RemoveNetworkClient),
 		router.NewRoute("POST", "/network/remove-clients", handlers.RemoveNetworkClients),
 		router.NewRoute("POST", "/network/provider-egress-location", handlers.ProviderEgressLocationSubmit),
+		router.NewRoute("GET", "/network/provider-egress-due", handlers.ProviderEgressLocationDue),
+		router.NewRoute("POST", "/network/provider-egress-attempt", handlers.ProviderEgressLocationAttempt),
 		router.NewRoute("GET", "/network/clients", handlers.NetworkClients),
 		router.NewRoute("GET", "/network/peers", handlers.NetworkPeers),
 		router.NewRoute("GET", "/network/provider-locations", handlers.NetworkGetProviderLocations),

api/handlers/provider_egress_location_handlers.go

@@ -5,12 +5,14 @@ import (
 	"encoding/json"
 	"io"
 	"net/http"
+	"strconv"
 	"sync"
 
 	"github.com/urnetwork/glog"
 
 	"github.com/urnetwork/server"
 	"github.com/urnetwork/server/controller"
+	"github.com/urnetwork/server/model"
 )
 
 // operatorSecretHeader carries the operator ingest secret. This endpoint is
@@ -91,3 +93,137 @@ func ProviderEgressLocationSubmit(w http.ResponseWriter, r *http.Request) {
 		glog.Infof("[pegl]could not write response. err = %s\n", err)
 	}
 }
+
+// ProviderEgressLocationAttempt records that the operator's prober tried to
+// probe a provider, whether or not the try produced a location.
+//
+// The prober reports a *failure* here; a success is reported by
+// ProviderEgressLocationSubmit above, whose provider_egress_location row
+// already defers the provider for the full staleness window. Reporting a
+// success here as well is harmless -- the attempt backoff is far shorter than
+// that window -- but redundant.
+//
+// This exists because ProviderEgressLocationDue would otherwise be starved by
+// providers that can never be probed successfully: they never get an egress
+// row, so they sort to the head of the queue on every poll forever. See
+// model.GetProviderEgressLocationDue.
+//
+// Same auth as the two endpoints around it: operator-to-server, the shared
+// secret header rather than a network jwt, fail-closed when the vault resource
+// is missing.
+func ProviderEgressLocationAttempt(w http.ResponseWriter, r *http.Request) {
+	secret := operatorIngestSecret()
+	provided := r.Header.Get(operatorSecretHeader)
+	if secret == "" || provided == "" || !hmac.Equal([]byte(secret), []byte(provided)) {
+		http.Error(w, "Unauthorized", http.StatusUnauthorized)
+		return
+	}
+
+	body, err := io.ReadAll(io.LimitReader(r.Body, maxProviderEgressLocationBody+1))
+	if err != nil {
+		http.Error(w, "Bad request", http.StatusBadRequest)
+		return
+	}
+	if len(body) > maxProviderEgressLocationBody {
+		http.Error(w, "Request too large", http.StatusRequestEntityTooLarge)
+		return
+	}
+
+	var args controller.RecordProviderEgressProbeAttemptArgs
+	if err := json.Unmarshal(body, &args); err != nil {
+		http.Error(w, "Bad request", http.StatusBadRequest)
+		return
+	}
+
+	result, err := controller.RecordProviderEgressProbeAttempt(r.Context(), &args)
+	if err != nil {
+		http.Error(w, err.Error(), http.StatusBadRequest)
+		return
+	}
+
+	w.Header().Set("Content-Type", "application/json")
+	if err := json.NewEncoder(w).Encode(result); err != nil {
+		glog.Infof("[pegl]could not write response. err = %s\n", err)
+	}
+}
+
+const (
+	// defaultProviderEgressDueLimit is the batch size when the caller does not
+	// ask for one.
+	defaultProviderEgressDueLimit = 100
+	// maxProviderEgressDueLimit bounds the batch size regardless of what the
+	// caller asks for, so one request cannot ask the database for the entire
+	// provider population.
+	maxProviderEgressDueLimit = 500
+)
+
+// providerEgressDueAge is how stale a stored probe must be before its provider
+// is offered up for re-probing. It is deliberately shorter than
+// model.ProviderEgressLocationMaxAge -- the age past which a stored location
+// stops being trusted at all. If the two were equal, every location would lapse
+// to the mmdb fallback at the exact moment it became due and stay lapsed until
+// the prober worked its way around to it; at half the max age the prober has a
+// full max-age/2 window to refresh a location before it expires.
+const providerEgressDueAge = model.ProviderEgressLocationMaxAge / 2
+
+// ProviderEgressLocationDueResult is the response body of
+// ProviderEgressLocationDue.
+type ProviderEgressLocationDueResult struct {
+	ClientIds []server.Id `json:"client_ids"`
+}
+
+// ProviderEgressLocationDue tells the operator's prober which providers to
+// probe next: those whose egress location has gone stale, and those that have
+// never been probed at all, oldest first.
+//
+// This moves the probe schedule from the prober's memory into the database.
+// The prober used to decide what to probe from an in-memory ttl cache, so a
+// restart re-probed the whole population and nothing durable recorded what was
+// actually due; observed_at already carries that information server-side, and
+// this exposes it.
+//
+// A provider is skipped if it has a fresh success *or* a recent attempt. The
+// second cutoff, ProviderEgressProbeAttemptBackoff, is much shorter than the
+// first: a provider that failed to probe should be retried within hours, but
+// must not be handed back on every poll, which is what would starve the rest of
+// the queue (see ProviderEgressLocationAttempt above).
+//
+// Same auth as ProviderEgressLocationSubmit above: operator-to-server, the
+// shared secret header rather than a network jwt, fail-closed when the vault
+// resource is missing.
+func ProviderEgressLocationDue(w http.ResponseWriter, r *http.Request) {
+	secret := operatorIngestSecret()
+	provided := r.Header.Get(operatorSecretHeader)
+	if secret == "" || provided == "" || !hmac.Equal([]byte(secret), []byte(provided)) {
+		http.Error(w, "Unauthorized", http.StatusUnauthorized)
+		return
+	}
+
+	limit := defaultProviderEgressDueLimit
+	if raw := r.URL.Query().Get("limit"); raw != "" {
+		parsed, err := strconv.Atoi(raw)
+		if err != nil || parsed < 1 {
+			// not clamped up to 1: `limit=0` would come back as an empty list,
+			// which the prober cannot distinguish from "nothing is due"
+			http.Error(w, "Bad request", http.StatusBadRequest)
+			return
+		}
+		limit = min(parsed, maxProviderEgressDueLimit)
+	}
+
+	// both cutoffs are computed here and passed as arguments; observed_at and
+	// attempt_at are naive timestamps holding utc, so comparing them to sql
+	// now() in the query would cast through the session timezone
+	now := server.NowUtc()
+	minObservedAt := now.Add(-providerEgressDueAge)
+	minAttemptAt := now.Add(-model.ProviderEgressProbeAttemptBackoff)
+
+	result := &ProviderEgressLocationDueResult{
+		ClientIds: model.GetProviderEgressLocationDue(r.Context(), minObservedAt, minAttemptAt, limit),
+	}
+
+	w.Header().Set("Content-Type", "application/json")
+	if err := json.NewEncoder(w).Encode(result); err != nil {
+		glog.Infof("[pegl]could not write response. err = %s\n", err)
+	}
+}