Trace source refresh decisions

docs/api.md

@@ -251,13 +251,50 @@ a passing fresh HTTPS/CONNECT check, and a fresh clean tamper check, with age
 bounded by `classification_max_age`. It is a point-in-time filter count, not a
 claim that the proxy will remain usable.
 
+## Operator source refresh traces
+
+`GET /report/source-refreshes` returns recent bounded source-decision traces.
+It is served on the admin listener with the other operator routes; when
+`admin_listen_addr` is configured it is absent from the public listener. When
+that listener is empty, this trace route accepts loopback connections only;
+keep ingress from forwarding operator paths. The endpoint is `no-store` and
+contains no source URLs, proxy endpoints, IP addresses, or credentials.
+
+Without `refresh_id`, it returns a bounded list of refresh summaries. `limit`
+defaults to 64 and is capped at 64. With `refresh_id=<32 lowercase hex
+characters>`, it returns the matching detailed trace. A trace links
+`refresh_id`, `generation`, `captured_at`, `started_at`, `completed_at`,
+`status` (`started`, `completed`, `failed`, or `cancelled`), and `result`.
+Per-source decisions include `selected`, `cooled`, `explored`, and
+`admission_cooled` booleans, overlapping reason codes, protection booleans for
+trusted priority and current-contract positive probes, plus bounded
+`fetch_status`/`merge_status` outcomes and timestamps. Fetch metrics are
+`fetch_attempts`,
+`http_attempts`, `network_bytes`, `fetch_duration_ms`, and `parsed`;
+`admitted_candidates` counts new candidate IDs actually accepted from that
+source during this refresh. Network bytes are response-body bytes, and fetch
+duration includes stream, parsing, and staging work rather than CPU time. They
+also include
+`admission_observation_status`
+(`not_selected`, `not_applicable`, `pending`, `recorded`, `censored`,
+`write_failed`, or `cancelled`) and `admission_observation_reasons`, which may
+include `truncated`, `candidate_cap`, `fetch_failed`, `merge_failed`,
+`cancelled`, `observation_write_failed`, `refresh_failed_before_fetch`, and
+`refresh_failed`. `admission_cooled` describes eligibility; it does not mean a
+source was actually omitted, and exploration can select a cooled source. A
+completed refresh can have a `partial` result when a selected
+source failed, was truncated or capped, or its admission observation write
+failed. Censored observations do not advance the full-pass zero-admission
+streak. Traces expire after 72 hours and at most 64 are retained. See the
+[Redis schema](redis-schema.md#source-refresh-traces).
+
 Admission refresh fields (`admission_refreshes`, `last_admitted`,
-`zero_admission_streak`, and `last_admission_at`) are updated only for a whole
-successful refresh pass and sources whose pass was not truncated or affected by
-the candidate cap. Failed, skipped, truncated, and capped sources do not add a
-zero-admission observation. `admitted` itself counts atomic new set insertions
-and may include accepted additions from a pass whose later step failed; it is
-not the same as a completed admission observation.
+`zero_admission_streak`, and `last_admission_at`) are updated only for eligible
+full-pass observations. Failed, skipped, truncated, capped, cancelled, or
+otherwise censored observations do not advance the zero-admission streak.
+`admitted` itself counts atomic new set insertions and may include accepted
+additions from a pass whose later step failed; it is not the same as a completed
+admission observation.
 
 Fetch cost fields sum HTTP client attempts, consumed HTTP response-body bytes,
 and elapsed stream work. Attempts count calls to the HTTP client; bytes exclude

docs/configuration.md

@@ -29,19 +29,22 @@ operate and have approval to use. Read the
 | `sources` | none (required) | Proxy feeds, such as `file:///data/proxies.txt` or an `https://` URL. An optional `scheme@` prefix sets the default scheme for bare `host:port` lines in that feed, for example `socks5@https://example.com/socks5.txt`. |
 | `default_proxy_scheme` | `http` | Scheme for bare `host:port` records whose feed has no `scheme@` prefix. |
 | `listen_addr` | `:8080` | Public HTTP listener. |
-| `admin_listen_addr` | empty | Optional second listener, for example `127.0.0.1:9090`, for `/metrics`, `/report`, `/report/sources`, and `/internal/api/v1/proxies`. When set, those endpoints are absent from `listen_addr`. Must differ from `listen_addr`. |
+| `admin_listen_addr` | empty | Optional second listener, for example `127.0.0.1:9090`, for `/metrics`, `/report`, `/report/sources`, `/report/source-refreshes`, and `/internal/api/v1/proxies`. When set, those endpoints are absent from `listen_addr`. Must differ from `listen_addr`. |
 | `public_base_url` | empty | Public origin of your deployment, such as `https://proxies.example.com`. Used for canonical, Open Graph and code-sample URLs on the built-in pages. When empty, samples show the origin the visitor reached and canonical tags are omitted. |
 | `trusted_proxy_cidrs` | none | CIDRs of ingress that strips caller-supplied forwarding headers and writes trusted ones. Leave empty to identify clients by socket peer. |
 | `internal_api_token_file` | empty | Bearer-token file for the internal API. The API is disabled when this is empty. See [deployment.md](deployment.md#internal-query-api). |
 | `prometheus_url` | empty | Prometheus base URL (`http(s)`, no credentials, query or fragment). The dashboard needs it. |
 | `public_max_limit` | `1000` (1–1000) | Largest page `/proxies` serves; a larger `limit` is clamped to it. An omitted `limit` is 100, or this value if smaller. Lower it to 100 to cap pages, after `freeproxyapi_public_large_page_requests_total` has stopped growing (see [deployment.md](deployment.md#lowering-the-public-page-size)). |
 
 `/metrics` exposes feed hostnames and probe settings. When `admin_listen_addr`
-is empty (the compatibility default), `/metrics`, `/report`, `/report/sources`
-and the internal API share the public listener, so your ingress must not forward
-them. The Compose example binds the admin listener to the container's loopback;
+is empty (the compatibility default), `/metrics`, `/report`, `/report/sources`,
+`/report/source-refreshes`, and the internal API share the public listener, so
+your ingress must not forward them. The Compose example binds the admin listener
+to the container's loopback;
 to scrape it, bind `0.0.0.0:9090` and publish the port only on a network you
 control.
+The source-refresh trace route also enforces loopback-only requests while
+`admin_listen_addr` is empty.
 
 ## Feed intake
 
@@ -88,7 +91,10 @@ IDs. It defaults to disabled (`0s`) and is bounded to 24 hours. The configured
 `source_learning_zero_admission_refreshes` threshold defaults to 3 and must be
 3–100. Only complete successful, uncapped and untruncated refresh passes count
 toward this guard; failed, skipped, truncated, or candidate-cap-affected passes
-are excluded. Sources at fallback priority 10 or better, sources with a positive
+are excluded, and censored observations do not advance the full-pass
+zero-admission streak. A completed refresh may report a partial result when a
+selected source fails, is truncated or capped, or its admission observation
+write fails. Sources at fallback priority 10 or better, sources with a positive
 observed probe outcome, and sources whose learned probe yield clears the
 configured minimum are protected. The one-in-ten exploration allowance remains
 in effect for cooled feeds. Any actual new candidate insertion clears that

docs/redis-schema.md

@@ -17,6 +17,9 @@ prefix. Every key below is relative to `namespace`, which defaults to
 | `<ns>:leased` | ZSET | Candidates currently leased across all replicas. Score is lease-expiry epoch milliseconds. |
 | `<ns>:validated` | SET | Candidate IDs whose `results10` history holds at least `min_samples_for_listing` outcomes (default 3) and whose rolling `ok_ratio_pct` is at least 80. Every completion, success or failure, re-evaluates that rule, so one lucky success does not list a candidate and one transient failure does not evict a stable one. Records without a ratio fall back to their latest outcome. |
 | `<ns>:source-refresh-lock` | STRING | Short-TTL `SETNX` lease so exactly one replica refreshes inventory sources at a time; the owner renews it while active and losing it is normal. |
+| `<ns>:source-refresh-trace:<refresh-id>` | STRING (gzip-compressed JSON) | Bounded detail trace for one source refresh, keyed by a 32-character lowercase hexadecimal opaque refresh ID. Includes refresh/generation timestamps, status/result, and bounded per-source decisions; TTL 72 hours. At most 64 traces are retained. |
+| `<ns>:source-refresh-trace-summary:<refresh-id>` | STRING (JSON) | Compact summary used by the bounded trace-list view. Retained and evicted with its detail trace; TTL 72 hours. |
+| `<ns>:source-refresh-traces` | ZSET | Bounded index of retained refresh trace IDs, ordered by capture time and trimmed to at most 64 entries. Trace storage is telemetry and fails open if Redis trace writes fail. |
 | `<ns>:source-stage:<token>` | ZSET or SET | Temporary staging keys for one source refresh, holding canonical endpoint strings. The shared refresh stage is a ZSET: score is the lowest source priority seen for the endpoint (`ZADD LT`), so it is both the cross-source dedupe index and the drain order. Each feed first stages into its own private SET (one `SADD` per record); only after the feed streamed completely is that SET `SPOP`-merged into the shared ZSET in bounded batches, so a feed that fails mid-stream contributes nothing. TTL 2 hours, re-armed on every staging batch, merge batch, and drain pass; every key is deleted when its refresh or feed finishes. |
 | `<ns>:source-quality:<source-id>` | HASH | Optional bounded feed-refresh and sampled probe aggregates. `source-id` is 32 lowercase hex characters derived from the exact feed URL and its effective parser scheme. TTL is 30 days. Used only when source learning is enabled. |
 | `<ns>:source-quality:active` | SET | The current configured source-ID catalog used to bound attribution and expiry. Replaced during refresh planning and expires after 24 hours. |
@@ -27,6 +30,27 @@ prefix. Every key below is relative to `namespace`, which defaults to
 | `<ns>:stats-leader` | STRING | Random token of the replica that holds the stats-leader lease. Taken with `SET NX PX` semantics and renewed on every stats publish (TTL `stats_leader_ttl`, default 90s); released on graceful shutdown. Only the holder scans the validated set. |
 | `<ns>:validated-aggregate` | STRING (JSON) | Validated-set aggregate written by the stats leader, only while its lease token matches: `computed_at_ms`, `by_country`, `stable_by_country`, `by_asn`, `by_anonymity`, `by_latency_band`, `stable`, `band_known`, `geo_mismatch`, and (when source learning is active) `by_source`, `routing_by_source`, `source_counts_observed`, `unattributed_validated`, and `unattributed_routing`. Source maps contain counts only, keyed by source ID. TTL `stats_aggregate_ttl` (default 5m). Followers ignore it once older than two publish intervals (60s) and scan locally instead. |
 
+## Source refresh traces
+
+Source refresh decision traces back the admin-only `GET /report/source-refreshes`
+endpoint (see [API reference](api.md#operator-source-refresh-traces)). Each
+trace links its opaque `refresh_id`, generation, capture/start/completion times,
+status, result, and per-source selection, cooldown, exploration, admission
+cooldown, protection, reason-code, fetch-status, and attempt-count fields.
+Per-source admission fields include `admission_observation_status`
+(`not_selected`, `not_applicable`, `pending`, `recorded`, `censored`,
+`write_failed`, or `cancelled`) and `admission_observation_reasons` (which may
+contain `truncated`, `candidate_cap`, `fetch_failed`, `merge_failed`,
+`cancelled`, `observation_write_failed`, `refresh_failed_before_fetch`, or
+`refresh_failed`).
+Reason codes may overlap. Admission cooldown records eligibility rather than
+actual omission, and an exploration decision may select a cooled source. Trace
+keys and their Redis index are TTL-backed and bounded to 64 retained records;
+trace telemetry is fail-open and does not determine refresh behavior. A refresh
+can complete with a partial result if a selected source failed, was truncated
+or capped, or its admission observation write failed. Censored observations do
+not advance the full-pass zero-admission streak.
+
 ## Source learning and quality history
 
 Source learning is opt-in (`source_learning_enabled`, default false; see

internal/monitor/health.go

@@ -461,6 +461,7 @@ func startHealthServer(addr string, runner *Runner) (*healthServer, error) {
 	registerAdmin := func(m *http.ServeMux) {
 		m.HandleFunc("/report", h.report)
 		m.HandleFunc("/report/sources", h.sourceReport)
+		m.HandleFunc("/report/source-refreshes", h.sourceRefreshes)
 		m.Handle("/metrics", withAccuracyMetrics(runner.metrics.Handler(), &runner.accuracy.metrics))
 		registerInternalAPI(m, runner, loadAPIToken(runner.config.InternalAPITokenFile))
 	}
@@ -1237,3 +1238,99 @@ func (h *healthHandlers) sourceReport(w http.ResponseWriter, r *http.Request) {
 	}
 	_ = json.NewEncoder(w).Encode(report)
 }
+
+// sourceRefreshes serves bounded, opaque source-decision traces through the
+// operator routes. Keep this path on the admin listener or behind ingress
+// rules that already exclude the other /report routes.
+func (h *healthHandlers) sourceRefreshes(w http.ResponseWriter, r *http.Request) {
+	if h.runner.config.AdminListenAddr == "" && !loopbackRemoteAddr(r.RemoteAddr) {
+		http.NotFound(w, r)
+		return
+	}
+	if r.Method != http.MethodGet && r.Method != http.MethodHead {
+		w.Header().Set("Allow", http.MethodGet+", "+http.MethodHead)
+		http.Error(w, "method not allowed", http.StatusMethodNotAllowed)
+		return
+	}
+	ctx, cancel := context.WithTimeout(r.Context(), readyzTimeout)
+	defer cancel()
+	w.Header().Set("Cache-Control", "no-store")
+
+	if refreshID := r.URL.Query().Get("refresh_id"); refreshID != "" {
+		if !validSourceRefreshTraceID(refreshID) {
+			writeJSON(w, r, http.StatusBadRequest, map[string]string{"status": "invalid_request"})
+			return
+		}
+		payload, err := h.runner.readStore().LoadSourceRefreshTrace(ctx, refreshID)
+		if errors.Is(err, store.ErrSourceRefreshTraceNotFound) {
+			writeJSON(w, r, http.StatusNotFound, map[string]string{"status": "not_found"})
+			return
+		}
+		if err != nil {
+			writeJSON(w, r, http.StatusServiceUnavailable, map[string]string{"status": "unavailable"})
+			return
+		}
+		trace, err := decodeSourceRefreshTrace(payload)
+		if err != nil || trace.RefreshID != refreshID {
+			writeJSON(w, r, http.StatusServiceUnavailable, map[string]string{"status": "unavailable"})
+			return
+		}
+		writeJSON(w, r, http.StatusOK, trace)
+		return
+	}
+
+	limit := store.MaxSourceRefreshTraces
+	if rawLimit := r.URL.Query().Get("limit"); rawLimit != "" {
+		parsed, err := strconv.Atoi(rawLimit)
+		if err != nil || parsed < 1 || parsed > store.MaxSourceRefreshTraces {
+			writeJSON(w, r, http.StatusBadRequest, map[string]string{"status": "invalid_request"})
+			return
+		}
+		limit = parsed
+	}
+	ids, err := h.runner.readStore().ListSourceRefreshTraceIDs(ctx, limit)
+	if err != nil {
+		writeJSON(w, r, http.StatusServiceUnavailable, map[string]string{"status": "unavailable"})
+		return
+	}
+	traces := make([]sourceRefreshTraceSummary, 0, len(ids))
+	for _, id := range ids {
+		payload, err := h.runner.readStore().LoadSourceRefreshTraceSummary(ctx, id)
+		if err != nil {
+			writeJSON(w, r, http.StatusServiceUnavailable, map[string]string{"status": "unavailable"})
+			return
+		}
+		summary, err := decodeSourceRefreshTraceSummary(payload, id)
+		if err != nil {
+			writeJSON(w, r, http.StatusServiceUnavailable, map[string]string{"status": "unavailable"})
+			return
+		}
+		traces = append(traces, summary)
+	}
+	writeJSON(w, r, http.StatusOK, sourceRefreshTraceList{
+		RetentionSeconds: int(store.SourceRefreshTraceRetention / time.Second),
+		MaxTraces:        store.MaxSourceRefreshTraces,
+		Traces:           traces,
+	})
+}
+
+func loopbackRemoteAddr(remote string) bool {
+	host, _, err := net.SplitHostPort(remote)
+	if err != nil {
+		return false
+	}
+	address, err := netip.ParseAddr(host)
+	return err == nil && address.IsLoopback()
+}
+
+func validSourceRefreshTraceID(id string) bool {
+	if len(id) != 32 {
+		return false
+	}
+	for _, char := range id {
+		if (char < '0' || char > '9') && (char < 'a' || char > 'f') {
+			return false
+		}
+	}
+	return true
+}