feat(prober): hourly blackhole sweep, and fetch the credential instead of hand-carrying it

README.md

@@ -45,26 +45,44 @@ go build ./cmd/egress-prober
 ./egress-prober \
   -api-url https://api.example.net \
   -platform-url wss://connect.example.net \
-  -by-jwt "$UR_PROBER_BY_JWT" \
   -operator-secret "$UR_OPERATOR_SECRET" \
   -concurrency 4 \
   -cache-ttl 24h \
   -interval 1h
 ```
 
+That run fetches the prober's own identity from the server. To supply one you
+provisioned yourself instead, add `-by-jwt "$UR_PROBER_BY_JWT"` (or export
+`UR_PROBER_BY_JWT`); it takes precedence and the fetch is skipped entirely.
+
 `-by-jwt` and `-operator-secret` may also be supplied via the
 `UR_PROBER_BY_JWT` and `UR_OPERATOR_SECRET` environment variables instead of
 flags, which is the recommended way to run this under systemd (keeps secrets
-out of `ps`/shell history). All four of `-api-url`, `-platform-url`, `-by-jwt`
-and `-operator-secret` are required; the prober exits immediately with a
-message naming the missing flag(s) if any are absent, rather than starting in
-a broken state.
-
-The prober needs its own network client identity (`-by-jwt`), provisioned like
-any other client. `-operator-secret` must match `ingest_secret` in the server's
-`provider_egress.yml` vault resource — it authenticates the pin fetch as well as
-ingest, so a wrong secret now stops the prober at startup rather than only
-having its submissions rejected.
+out of `ps`/shell history). `-api-url`, `-platform-url` and `-operator-secret`
+are required; the prober exits immediately with a message naming the missing
+flag(s) if any are absent, rather than starting in a broken state.
+
+The prober needs its own network client identity (`-by-jwt`). **Leave it empty
+and the prober fetches one for itself** from the server's
+`/network/prober-credential` endpoint, authenticating with `-operator-secret` —
+no hand-provisioned identity, and one less secret to place. The server mints
+that identity in a bootstrap task which runs every 6h, so a prober brought up
+alongside a fresh deployment may start before its credential exists: it waits
+for it, logging one line per attempt on a backoff capped at 5 minutes, rather
+than exiting into a restart loop. The wait has no deadline of its own — impose
+one with the supervisor's start timeout if a deployment wants it.
+
+An explicitly supplied `-by-jwt` (or `UR_PROBER_BY_JWT`) always wins and the
+endpoint is never contacted, so an existing deployment that provisions the
+identity by hand is unaffected and acquires no dependency on it. Either way the
+jwt goes through the same startup check, so a credential the process cannot use
+stops it at startup instead of leaving a prober that looks healthy and probes
+nothing.
+
+`-operator-secret` must match `ingest_secret` in the server's
+`provider_egress.yml` vault resource — it authenticates the credential fetch and
+the pin fetch as well as ingest, so a wrong secret stops the prober at startup
+rather than only having its submissions rejected.
 
 The server must have observed the geolocation certificate pins before the
 prober can start: it fetches them at startup and **refuses to run without a

cmd/egress-prober/blackhole.go

@@ -0,0 +1,220 @@
+package main
+
+import (
+	"context"
+	"errors"
+	"fmt"
+	"log"
+	"sync"
+	"time"
+
+	"github.com/urnetwork/connect"
+
+	"github.com/urnetwork/operator-proxy/egresshealth"
+	"github.com/urnetwork/operator-proxy/ingest"
+	"github.com/urnetwork/operator-proxy/providertunnel"
+)
+
+// blackholeSweeper runs the cheap liveness check across the whole fleet on its
+// own cadence, independently of the geolocation/health pass.
+//
+// The two must not share a loop. The full pass spends minutes per provider and
+// sweeps the fleet over hours to days; in that window a provider that silently
+// stops forwarding keeps its last passing measurement and stays in the public
+// list. This exists to close that window, which only works if it runs on its
+// own much shorter one.
+type blackholeSweeper struct {
+	operator    *ingest.Client
+	tunnelCfg   providertunnel.Config
+	pins        *pinSet
+	timeout     time.Duration
+	concurrency int
+	limit       int
+}
+
+// maxBlackholeRounds bounds one sweep's batches. 40 rounds x the server's 5000
+// ceiling is far above any real fleet, so it never truncates a legitimate
+// sweep; it exists so a server that keeps handing back work cannot hold a pass
+// open indefinitely and starve the interval.
+const maxBlackholeRounds = 40
+
+// blackholeResult carries one provider's outcome out of the worker pool.
+type blackholeResult struct {
+	check   ingest.BlackholeCheck
+	dark    bool
+	tunnel  bool
+	details string
+}
+
+// sweep runs one pass: ask what is due, check each, report the batch.
+//
+// Returns the number checked, so the caller can tell "the fleet is covered"
+// from "the queue handed us nothing", which look identical in a log line.
+func (s *blackholeSweeper) sweep(ctx context.Context) (checked int, err error) {
+	clientIds, err := s.operator.BlackholeDue(ctx, s.limit)
+	if err != nil {
+		return 0, err
+	}
+	if len(clientIds) == 0 {
+		return 0, nil
+	}
+
+	results := make([]blackholeResult, len(clientIds))
+
+	sem := make(chan struct{}, s.concurrency)
+	var wg sync.WaitGroup
+	for i, clientId := range clientIds {
+		if ctx.Err() != nil {
+			break
+		}
+		wg.Add(1)
+		go func(i int, clientId string) {
+			defer wg.Done()
+			sem <- struct{}{}
+			defer func() { <-sem }()
+			results[i] = s.checkOne(ctx, clientId)
+		}(i, clientId)
+	}
+	wg.Wait()
+
+	checks := make([]ingest.BlackholeCheck, 0, len(results))
+	var dark, tunnelFailed int
+	for _, r := range results {
+		if r.check.ClientId == "" {
+			// never ran: the pass was cancelled before this slot started
+			continue
+		}
+		checks = append(checks, r.check)
+		if r.dark {
+			dark++
+			if r.tunnel {
+				tunnelFailed++
+			}
+			log.Printf("blackhole: provider=%s DARK %s", r.check.ClientId, r.details)
+		}
+	}
+
+	if len(checks) == 0 {
+		return 0, nil
+	}
+	if err := s.operator.SubmitBlackholeChecks(ctx, checks); err != nil {
+		// the whole batch is lost, not part of it -- the server validates before
+		// writing -- so say how much
+		return 0, fmt.Errorf("submitting %d checks: %w", len(checks), err)
+	}
+
+	log.Printf("blackhole: pass checked=%d dark=%d (tunnel_failed=%d) ok=%d",
+		len(checks), dark, tunnelFailed, len(checks)-dark)
+	return len(checks), nil
+}
+
+// checkOne opens a tunnel through one provider and asks whether anything gets
+// through.
+//
+// A tunnel that cannot be opened counts as dark, and that is a deliberate
+// choice rather than an oversight: from a client's point of view a provider it
+// cannot establish a circuit through is exactly as useless as one that carries
+// nothing, and the whole purpose of this signal is to stop advertising
+// providers a client cannot use. It is recorded under its own failure class so
+// the two remain distinguishable in the data.
+func (s *blackholeSweeper) checkOne(ctx context.Context, clientId string) blackholeResult {
+	checkedAt := time.Now().UTC()
+
+	id, err := connect.ParseId(clientId)
+	if err != nil {
+		return blackholeResult{
+			check:   ingest.BlackholeCheck{ClientId: clientId, OK: false, Failure: "bad_client_id", CheckedAt: checkedAt},
+			dark:    true,
+			details: err.Error(),
+		}
+	}
+
+	cfg := s.tunnelCfg
+	cfg.Pins = s.pins.get()
+	t, err := providertunnel.Open(ctx, cfg, id)
+	if err != nil {
+		return blackholeResult{
+			check:   ingest.BlackholeCheck{ClientId: clientId, OK: false, Failure: "tunnel_failed", CheckedAt: checkedAt},
+			dark:    true,
+			tunnel:  true,
+			details: err.Error(),
+		}
+	}
+	defer t.Close()
+
+	// Only the blackhole destinations are allowed through this client. The full
+	// pass allows the whole egress-health table and the bandwidth targets; this
+	// check reaches three connectivity endpoints and nothing else, so the
+	// allowlist says exactly that.
+	client := t.HTTPClientForHosts(s.timeout, egresshealth.BlackholeHosts())
+
+	res := egresshealth.Blackhole(ctx, client, egresshealth.Options{PerRequestTimeout: s.timeout})
+
+	check := ingest.BlackholeCheck{ClientId: clientId, OK: res.OK, CheckedAt: checkedAt}
+	if !res.OK {
+		check.Failure = res.Failure
+	}
+
+	details := ""
+	if !res.OK {
+		for _, r := range res.Results {
+			details += r.Name + "=" + r.Err + " "
+		}
+	}
+	return blackholeResult{check: check, dark: !res.OK, details: details}
+}
+
+// run sweeps on the interval until the context ends.
+//
+// A pass that errors is logged and retried on the next tick rather than
+// stopping the sweeper: the server being briefly unreachable must not silently
+// end blackhole detection for the life of the process. ErrBlackholeUnsupported
+// is the one exception -- an older server will never grow the endpoint mid-run,
+// so it says so once and stops instead of logging the same 404 hourly.
+func (s *blackholeSweeper) run(ctx context.Context, interval time.Duration) {
+	for {
+		start := time.Now()
+		// Drain the queue, do not take one batch and sleep. The requirement is
+		// that the WHOLE fleet is checked every interval, and the batch size is the
+		// server's per-request ceiling, not the size of the fleet: at 500 per
+		// request against ~2,700 eligible providers, one batch per hour covers
+		// under a fifth of them and the oldest evidence would age out faster
+		// than the sweep reaches it. Rounds are bounded so a server that keeps
+		// returning work cannot hold a pass open forever.
+		total, err := 0, error(nil)
+		for round := 0; round < maxBlackholeRounds; round++ {
+			var checked int
+			checked, err = s.sweep(ctx)
+			total += checked
+			if err != nil || checked == 0 || ctx.Err() != nil {
+				break
+			}
+		}
+		checked := total
+		switch {
+		case err == nil:
+			if checked == 0 {
+				log.Printf("blackhole: pass found nothing due")
+			} else {
+				log.Printf("blackhole: sweep complete: %d checked in %s", checked, time.Since(start).Round(time.Second))
+			}
+		case errors.Is(err, ingest.ErrBlackholeUnsupported):
+			log.Printf("blackhole: the server does not implement the blackhole endpoints; sweeping is disabled")
+			return
+		case errors.Is(err, ingest.ErrUnauthorized):
+			// same posture as the credential self-check: a rejected secret is a
+			// broken deployment, and retrying hourly would hide it behind a
+			// sweep that never records anything
+			log.Printf("blackhole: the server rejected the operator secret; sweeping is disabled. Fix -operator-secret and restart.")
+			return
+		default:
+			log.Printf("blackhole: pass failed after %s: %s", time.Since(start).Round(time.Second), err)
+		}
+
+		select {
+		case <-ctx.Done():
+			return
+		case <-time.After(interval):
+		}
+	}
+}