diff --git a/docs/api.md b/docs/api.md index a02eb6c..f29fbef 100644 --- a/docs/api.md +++ b/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 diff --git a/docs/configuration.md b/docs/configuration.md index 6304072..5c70367 100644 --- a/docs/configuration.md +++ b/docs/configuration.md @@ -29,7 +29,7 @@ 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). | @@ -37,11 +37,14 @@ operate and have approval to use. Read the | `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 diff --git a/docs/redis-schema.md b/docs/redis-schema.md index 960b195..98bef9a 100644 --- a/docs/redis-schema.md +++ b/docs/redis-schema.md @@ -17,6 +17,9 @@ prefix. Every key below is relative to `namespace`, which defaults to | `:leased` | ZSET | Candidates currently leased across all replicas. Score is lease-expiry epoch milliseconds. | | `: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. | | `: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. | +| `:source-refresh-trace:` | 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. | +| `:source-refresh-trace-summary:` | STRING (JSON) | Compact summary used by the bounded trace-list view. Retained and evicted with its detail trace; TTL 72 hours. | +| `: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. | | `:source-stage:` | 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. | | `:source-quality:` | 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. | | `: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 | `: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. | | `: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 diff --git a/internal/monitor/health.go b/internal/monitor/health.go index 65aebe9..3383989 100644 --- a/internal/monitor/health.go +++ b/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 +} diff --git a/internal/monitor/metrics.go b/internal/monitor/metrics.go index 50dce5a..2d025bd 100644 --- a/internal/monitor/metrics.go +++ b/internal/monitor/metrics.go @@ -67,32 +67,33 @@ type targetProbeMetrics struct { // outcomes, and global-budget denials. Target-labeled probe metrics expose only // the configured scheme, host, and path (never query strings or credentials). type Metrics struct { - sourceRefreshTotal [4]atomic.Int64 // ok, error, skipped, canceled - sourceFailures sourceFailureMetrics - sourceRecordsAdded atomic.Int64 - sourceRecordsReject atomic.Int64 - sourcesTruncated atomic.Int64 - candidatesCapped atomic.Int64 - sourceFetchFailures atomic.Int64 - sourceFetchRetries atomic.Int64 - sourceFetchRecovered atomic.Int64 - claimsTotal atomic.Int64 - claimBatches atomic.Int64 - claimEmpty atomic.Int64 - reclaimsTotal atomic.Int64 - probeResultsTotal [2]atomic.Int64 - probeUploadBytes atomic.Int64 - probeDownloadBytes atomic.Int64 - budgetDenialsTotal atomic.Int64 - resultConflictsTotal atomic.Int64 - evictionsTotal atomic.Int64 - discardedTotal atomic.Int64 - sourceTombstoned atomic.Int64 - internalAPIFailures atomic.Int64 - publicQueryFailures atomic.Int64 - publicStaleResponses atomic.Int64 - publicLargePageRequests atomic.Int64 - publicRefreshFailures atomic.Int64 + sourceRefreshTotal [4]atomic.Int64 // ok, error, skipped, canceled + sourceRefreshTraceWriteFailures atomic.Int64 + sourceFailures sourceFailureMetrics + sourceRecordsAdded atomic.Int64 + sourceRecordsReject atomic.Int64 + sourcesTruncated atomic.Int64 + candidatesCapped atomic.Int64 + sourceFetchFailures atomic.Int64 + sourceFetchRetries atomic.Int64 + sourceFetchRecovered atomic.Int64 + claimsTotal atomic.Int64 + claimBatches atomic.Int64 + claimEmpty atomic.Int64 + reclaimsTotal atomic.Int64 + probeResultsTotal [2]atomic.Int64 + probeUploadBytes atomic.Int64 + probeDownloadBytes atomic.Int64 + budgetDenialsTotal atomic.Int64 + resultConflictsTotal atomic.Int64 + evictionsTotal atomic.Int64 + discardedTotal atomic.Int64 + sourceTombstoned atomic.Int64 + internalAPIFailures atomic.Int64 + publicQueryFailures atomic.Int64 + publicStaleResponses atomic.Int64 + publicLargePageRequests atomic.Int64 + publicRefreshFailures atomic.Int64 candidatesGauge atomic.Int64 sourceUniqueGauge atomic.Int64 @@ -238,6 +239,10 @@ func (m *Metrics) RecordSourceRefresh(result string) { m.sourceRefreshTotal[index].Add(1) } +func (m *Metrics) AddSourceRefreshTraceWriteFailure() { + m.sourceRefreshTraceWriteFailures.Add(1) +} + func (m *Metrics) AddSourceRecords(added, rejected int64) { if added > 0 { m.sourceRecordsAdded.Add(added) @@ -520,6 +525,7 @@ func (m *Metrics) Handler() http.Handler { {`freeproxyapi_source_refreshes_total{result="error"}`, "", "", m.sourceRefreshTotal[1].Load}, {`freeproxyapi_source_refreshes_total{result="skipped"}`, "", "", m.sourceRefreshTotal[2].Load}, {`freeproxyapi_source_refreshes_total{result="canceled"}`, "", "", m.sourceRefreshTotal[3].Load}, + {"freeproxyapi_source_refresh_trace_write_failures_total", "Source-refresh decision trace writes that failed open without interrupting refresh work.", "counter", m.sourceRefreshTraceWriteFailures.Load}, {"freeproxyapi_source_records_added_total", "Inventory records accepted into the candidate set.", "counter", m.sourceRecordsAdded.Load}, {"freeproxyapi_source_records_rejected_total", "Inventory records rejected at parse time.", "counter", m.sourceRecordsReject.Load}, {"freeproxyapi_sources_truncated_total", "Feeds cut off at max_records_per_source during a refresh.", "counter", m.sourcesTruncated.Load}, diff --git a/internal/monitor/runner.go b/internal/monitor/runner.go index 6485be2..8bbec7e 100644 --- a/internal/monitor/runner.go +++ b/internal/monitor/runner.go @@ -619,6 +619,8 @@ func (r *Runner) refresh(ctx context.Context) (err error) { return sourceLockBusyError{} } refreshCtx, stopHeartbeat := r.startSourceLockHeartbeat(ctx, token) + r.beginSourceRefreshTrace(refreshCtx) + defer func() { r.finishSourceRefreshTrace(refreshCtx, ctx, err) }() startedAt := time.Now() defer func() { // Stop renewing before changing the final lease state, so a tick that @@ -649,6 +651,9 @@ func (r *Runner) refresh(ctx context.Context) (err error) { r.recordRefreshError() return err } + if !r.config.SourceLearningEnabled { + r.setLegacySourceRefreshTracePlan(refreshCtx, r.config.SourceSpecs) + } if selected == 0 { r.metrics.RecordSourceRefresh("skipped") return nil @@ -721,6 +726,7 @@ func (r *Runner) drainSourceRefresh(ctx context.Context, stageKey string, result mergeErr error ) for result := range results { + r.recordSourceRefreshFetchResult(ctx, result) if result.err != nil { if result.stage == sourceStageMerge { // A failed merge leaves the shared stage partially populated @@ -786,7 +792,9 @@ func (r *Runner) drainSourceRefresh(ctx context.Context, stageKey string, result if r.config.SourceLearningEnabled { _, index, decodeErr := store.DecodeSourceStageScore(priority) if decodeErr == nil && index >= 0 && index < len(r.config.SourceSpecs) { - admissionExcluded[r.sourceID(r.config.SourceSpecs[index])] = true + sourceID := r.sourceID(r.config.SourceSpecs[index]) + admissionExcluded[sourceID] = true + r.recordSourceRefreshAdmissionCensored(sourceID, "candidate_cap") } } totalCapped += len(batch) @@ -802,7 +810,9 @@ func (r *Runner) drainSourceRefresh(ctx context.Context, stageKey string, result if err != nil || index < 0 || index >= len(r.config.SourceSpecs) { return fmt.Errorf("invalid source admission score") } - admissions[r.sourceID(r.config.SourceSpecs[index])] += int64(counts.Added) + sourceID := r.sourceID(r.config.SourceSpecs[index]) + admissions[sourceID] += int64(counts.Added) + r.recordSourceRefreshAdmissionDelta(sourceID, int64(counts.Added)) } if room > 0 { if room -= int64(counts.Added); room < 0 { @@ -832,11 +842,15 @@ func (r *Runner) drainSourceRefresh(ctx context.Context, stageKey string, result sort.Strings(ids) for start := 0; start < len(ids); start += store.MaxSourceQualityReadBatch { batch := make(map[string]int64) - for _, id := range ids[start:min(start+store.MaxSourceQualityReadBatch, len(ids))] { + batchIDs := ids[start:min(start+store.MaxSourceQualityReadBatch, len(ids))] + for _, id := range batchIDs { batch[id] = admissions[id] } if err := r.store.RecordSourceAdmissionRefresh(ctx, batch, r.nowFunc()); err != nil { log.Printf("source admission observation failed count=%d", len(batch)) + r.recordSourceRefreshAdmissionRecorded(batchIDs, true) + } else { + r.recordSourceRefreshAdmissionRecorded(batchIDs, false) } } } @@ -941,12 +955,16 @@ func sourceQueueTime(now time.Time, priority int64) time.Time { } type sourceRefreshResult struct { - spec SourceSpec - stage string - parsed int - rejected int - err error - truncated bool + spec SourceSpec + stage string + parsed int + rejected int + err error + truncated bool + fetchAttempts int + httpAttempts int64 + networkBytes int64 + fetchDurationMS int64 } const ( @@ -972,8 +990,18 @@ var errSourceRecordLimit = errors.New("source record limit reached") // shared ZSET with ZADD LT, touching each record once instead of staging it // into a set+zset pair twice and removing it from both. func (r *Runner) refreshSource(ctx context.Context, spec SourceSpec, stageKey string) sourceRefreshResult { + r.recordSourceRefreshFetchStarted(spec) + var httpAttempts int64 + var networkBytes, fetchDurationMS int64 for attempt := 1; attempt <= sourceFetchAttempts; attempt++ { result := r.refreshSourceAttempt(ctx, spec, stageKey) + result.fetchAttempts = attempt + httpAttempts += result.httpAttempts + result.httpAttempts = httpAttempts + networkBytes += result.networkBytes + result.networkBytes = networkBytes + fetchDurationMS += result.fetchDurationMS + result.fetchDurationMS = fetchDurationMS if result.err == nil { if attempt > 1 { r.metrics.AddSourceFetchRecovered(1) @@ -1049,6 +1077,9 @@ func (r *Runner) refreshSourceAttempt(ctx context.Context, spec SourceSpec, stag return flush() }) err = fetchErr + result.httpAttempts = fetchStats.HTTPAttempts + result.networkBytes = fetchStats.NetworkBytes + result.fetchDurationMS = fetchStats.Duration.Milliseconds() result.truncated = truncated if r.config.SourceLearningEnabled { costCtx, cancel := context.WithTimeout(context.WithoutCancel(ctx), releaseTimeout) diff --git a/internal/monitor/source_learning.go b/internal/monitor/source_learning.go index 624f651..8120a8a 100644 --- a/internal/monitor/source_learning.go +++ b/internal/monitor/source_learning.go @@ -31,6 +31,9 @@ type SourceLearningCandidate struct { ID string FallbackPriority int64 Quality store.SourceQuality + // ProbeContractCurrent is set by the refresh snapshot loader after it + // invalidates probe observations recorded under an older contract. + ProbeContractCurrent bool } // RankedSource is one entry selected for the next source refresh. Priority is @@ -46,12 +49,36 @@ type RankedSource struct { AdmissionCooled bool } +// SourceLearningDecision records the planner's actual decision for one source. +// Reasons retain overlapping cooldowns; an explored source can therefore be +// selected while still reporting each cooldown that exploration bypassed. +type SourceLearningDecision struct { + ID string + Selected bool + Cooled bool + Explored bool + AdmissionCooled bool + TrustedPriorityProtected bool + CurrentPositiveProbeProtected bool + ProbeContractCurrent bool + PlanPosition *int + Reasons []string +} + +// DetailedSourceLearningPlan pairs the sources the refresh will actually +// fetch with a decision record for every unique configured source identity. +type DetailedSourceLearningPlan struct { + Selected []RankedSource + Decisions []SourceLearningDecision +} + type scoredSource struct { candidate SourceLearningCandidate score float64 learned bool cooled bool admissionCooled bool + cooldownReasons []string } // StableSourceID shares store's URL-plus-scheme identity so the refresh planner @@ -77,8 +104,15 @@ func SourceYield(q store.SourceQuality, now time.Time, config SourceLearningConf // one-in-N exploration refresh, so a bad early estimate cannot suppress a feed // forever. With learning disabled, fallback priorities are returned unchanged. func BuildSourceLearningPlan(candidates []SourceLearningCandidate, refreshNumber uint64, now time.Time, config SourceLearningConfig) []RankedSource { + return BuildSourceLearningPlanDetailed(candidates, refreshNumber, now, config).Selected +} + +// BuildSourceLearningPlanDetailed is the source planner used by refreshes and +// trace reporting. It returns the same ranked plan as BuildSourceLearningPlan +// plus one exact, reason-preserving decision for every unique candidate. +func BuildSourceLearningPlanDetailed(candidates []SourceLearningCandidate, refreshNumber uint64, now time.Time, config SourceLearningConfig) DetailedSourceLearningPlan { if len(candidates) == 0 { - return nil + return DetailedSourceLearningPlan{} } minimum, halfLife := config.normalized() seen := make(map[string]struct{}, len(candidates)) @@ -96,20 +130,11 @@ func BuildSourceLearningPlan(candidates []SourceLearningCandidate, refreshNumber score, samples = 0.5, 0 } eligible := config.Enabled && !candidate.Quality.Corrupt && samples >= minimum - cooled := false - admissionCooled := false - if config.Enabled && !candidate.Quality.Corrupt { - cooled = candidate.Quality.CooldownUntil.After(now) - if eligible && config.LowYieldThreshold > 0 && score < config.LowYieldThreshold && config.LowYieldCooldown > 0 { - lowYieldUntil := candidate.Quality.ProbeUpdatedAt.Add(config.LowYieldCooldown) - cooled = cooled || lowYieldUntil.After(now) - } - admissionCooled = shouldCoolZeroAdmission(candidate, now, config) - cooled = cooled || admissionCooled - } + cooldownReasons := sourceCooldownReasons(candidate, now, config, eligible, score) + admissionCooled := shouldCoolZeroAdmission(candidate, now, config) scored = append(scored, scoredSource{ - candidate: candidate, score: score, learned: eligible, cooled: cooled, - admissionCooled: admissionCooled, + candidate: candidate, score: score, learned: eligible, cooled: len(cooldownReasons) > 0, + admissionCooled: admissionCooled, cooldownReasons: cooldownReasons, }) } if !config.Enabled { @@ -120,10 +145,13 @@ func BuildSourceLearningPlan(candidates []SourceLearningCandidate, refreshNumber return false // stable sort preserves configured source order on ties }) plan := make([]RankedSource, 0, len(scored)) + decisions := make([]SourceLearningDecision, 0, len(scored)) for _, item := range scored { plan = append(plan, RankedSource{ID: item.candidate.ID, Priority: item.candidate.FallbackPriority}) + position := len(plan) - 1 + decisions = append(decisions, sourceLearningDecision(item, true, false, &position, []string{"learning_disabled"})) } - return plan + return DetailedSourceLearningPlan{Selected: plan, Decisions: decisions} } interval := config.ExplorationInterval @@ -174,11 +202,14 @@ func BuildSourceLearningPlan(candidates []SourceLearningCandidate, refreshNumber exploring = exploring[:maxExploration] } plan := make([]RankedSource, 0, len(active)+len(exploring)) + selectedByID := make(map[string]RankedSource, len(active)+len(exploring)) appendSource := func(item scoredSource, isExploration bool) { - plan = append(plan, RankedSource{ + ranked := RankedSource{ ID: item.candidate.ID, Priority: int64(len(plan)), Score: item.score, Learned: item.learned, Exploration: isExploration, AdmissionCooled: item.admissionCooled, - }) + } + plan = append(plan, ranked) + selectedByID[item.candidate.ID] = ranked } exploreIndex := 0 for i, item := range active { @@ -191,7 +222,63 @@ func BuildSourceLearningPlan(candidates []SourceLearningCandidate, refreshNumber for ; exploreIndex < len(exploring); exploreIndex++ { appendSource(exploring[exploreIndex], true) } - return plan + decisions := make([]SourceLearningDecision, 0, len(scored)) + for _, item := range scored { + ranked, selected := selectedByID[item.candidate.ID] + var position *int + if selected { + value := int(ranked.Priority) + position = &value + } + reasons := append([]string(nil), item.cooldownReasons...) + if ranked.Exploration { + reasons = append(reasons, "exploration") + } + if len(item.cooldownReasons) == 0 { + if item.candidate.Quality.Corrupt { + reasons = append(reasons, "quality_corrupt_fail_open") + } else { + reasons = append(reasons, "active") + } + } + decisions = append(decisions, sourceLearningDecision(item, selected, ranked.Exploration, position, reasons)) + } + return DetailedSourceLearningPlan{Selected: plan, Decisions: decisions} +} + +func sourceLearningDecision(item scoredSource, selected, explored bool, position *int, reasons []string) SourceLearningDecision { + quality := item.candidate.Quality + return SourceLearningDecision{ + ID: item.candidate.ID, + Selected: selected, + Cooled: item.cooled, + Explored: explored, + AdmissionCooled: item.admissionCooled, + TrustedPriorityProtected: item.candidate.FallbackPriority <= 10, + CurrentPositiveProbeProtected: item.candidate.ProbeContractCurrent && (quality.ProbePasses > 0 || quality.ProbeSuccessfulCompletions > 0), + ProbeContractCurrent: item.candidate.ProbeContractCurrent, + PlanPosition: position, + Reasons: reasons, + } +} + +func sourceCooldownReasons(candidate SourceLearningCandidate, now time.Time, config SourceLearningConfig, eligible bool, score float64) []string { + if !config.Enabled || candidate.Quality.Corrupt { + return nil + } + var reasons []string + if candidate.Quality.CooldownUntil.After(now) { + reasons = append(reasons, "failure_cooldown") + } + if eligible && config.LowYieldThreshold > 0 && score < config.LowYieldThreshold && config.LowYieldCooldown > 0 { + if candidate.Quality.ProbeUpdatedAt.Add(config.LowYieldCooldown).After(now) { + reasons = append(reasons, "low_yield_cooldown") + } + } + if shouldCoolZeroAdmission(candidate, now, config) { + reasons = append(reasons, "zero_admission_cooldown") + } + return reasons } func shouldCoolZeroAdmission(candidate SourceLearningCandidate, now time.Time, config SourceLearningConfig) bool { diff --git a/internal/monitor/source_learning_runner.go b/internal/monitor/source_learning_runner.go index 04ca93c..5b30c28 100644 --- a/internal/monitor/source_learning_runner.go +++ b/internal/monitor/source_learning_runner.go @@ -39,9 +39,11 @@ type sourceLearningReport struct { AdmissionCooled int `json:"admission_cooled"` } type sourcePlanState struct { - mu sync.RWMutex - feeds map[string]plannedSource - report sourceLearningReport + mu sync.RWMutex + feeds map[string]plannedSource + report sourceLearningReport + trace *sourceRefreshTrace + traceDecisionIndex map[string]int } func (r *Runner) sourceID(spec SourceSpec) string { @@ -66,6 +68,8 @@ func (r *Runner) prepareSourceLearning(ctx context.Context) (int, error) { return 0, err } quality := make(map[string]store.SourceQuality, len(ids)) + probeContractCurrent := make(map[string]bool, len(ids)) + currentContract := r.sourceProbeContract() now := r.nowFunc() for start := 0; start < len(ids); start += store.MaxSourceQualityReadBatch { batch, err := r.store.LoadSourceQuality(ctx, ids[start:min(start+store.MaxSourceQualityReadBatch, len(ids))], now) @@ -73,7 +77,8 @@ func (r *Runner) prepareSourceLearning(ctx context.Context) (int, error) { return 0, err } for id, q := range batch { - if q.ProbeContract != r.sourceProbeContract() { + probeContractCurrent[id] = q.ProbeContract == currentContract && validSourceProbeContract(currentContract) + if q.ProbeContract != currentContract { q.ProbeSamples = 0 q.ProbePasses = 0 q.ProbeUpdatedAt = time.Time{} @@ -89,9 +94,13 @@ func (r *Runner) prepareSourceLearning(ctx context.Context) (int, error) { if err != nil { return 0, err } + r.setSourceRefreshTraceGeneration(ctx, generation, now, len(ids)) candidates := make([]SourceLearningCandidate, 0, len(ids)) for _, id := range ids { - candidates = append(candidates, SourceLearningCandidate{ID: id, FallbackPriority: sourcePriority(r.config.SourceSpecs[indexes[id]]), Quality: quality[id]}) + candidates = append(candidates, SourceLearningCandidate{ + ID: id, FallbackPriority: sourcePriority(r.config.SourceSpecs[indexes[id]]), + Quality: quality[id], ProbeContractCurrent: probeContractCurrent[id], + }) } learningConfig := SourceLearningConfig{ Enabled: true, MinSamples: 8, HalfLife: 7 * 24 * time.Hour, @@ -99,11 +108,12 @@ func (r *Runner) prepareSourceLearning(ctx context.Context) (int, error) { ZeroAdmissionCooldown: r.config.SourceLearningZeroAdmissionCooldown, ZeroAdmissionRefreshes: int64(r.config.SourceLearningZeroAdmissionRefreshes), } - plan := BuildSourceLearningPlan(candidates, generation, now, learningConfig) + detailedPlan := BuildSourceLearningPlanDetailed(candidates, generation, now, learningConfig) + plan := detailedPlan.Selected feeds := make(map[string]plannedSource, len(plan)) report := sourceLearningReport{Enabled: true, Prepared: true, Generation: generation, Configured: len(ids), Selected: len(plan), Skipped: len(ids) - len(plan)} - for _, candidate := range candidates { - if shouldCoolZeroAdmission(candidate, now, learningConfig) { + for _, decision := range detailedPlan.Decisions { + if decision.AdmissionCooled { report.AdmissionCooled++ } } @@ -124,6 +134,7 @@ func (r *Runner) prepareSourceLearning(ctx context.Context) (int, error) { r.sourcePlan.mu.Lock() r.sourcePlan.feeds, r.sourcePlan.report = feeds, report r.sourcePlan.mu.Unlock() + r.setSourceRefreshTracePlan(ctx, generation, now, len(ids), detailedPlan.Decisions) return len(plan), nil } diff --git a/internal/monitor/source_learning_test.go b/internal/monitor/source_learning_test.go index 6cf73d3..6c8ebdf 100644 --- a/internal/monitor/source_learning_test.go +++ b/internal/monitor/source_learning_test.go @@ -336,3 +336,80 @@ func TestCorruptQualityFailsOpenForEveryLearningCooldown(t *testing.T) { t.Fatalf("corrupt source did not fail open at fallback priority: %+v", plan) } } + +func TestDetailedSourcePlanPreservesActualDecisionReasonsAndProtections(t *testing.T) { + now := time.Unix(1_800_000_000, 0) + exploredID := fmt.Sprintf("%032x", 1) + omittedID := fmt.Sprintf("%032x", 2) + zeroQuality := store.SourceQuality{ + ProbeSamples: 8, ProbePasses: 0, ProbeUpdatedAt: now, + CooldownUntil: now.Add(10 * time.Minute), + AdmissionRefreshes: 3, ZeroAdmissionStreak: 3, LastAdmissionAt: now.Add(-time.Minute), + } + config := SourceLearningConfig{ + Enabled: true, MinSamples: 8, LowYieldThreshold: 0.2, LowYieldCooldown: 24 * time.Hour, + ZeroAdmissionCooldown: time.Hour, ZeroAdmissionRefreshes: 3, ExplorationInterval: 10, + } + var generation uint64 + for generation = 1; generation < 1000; generation++ { + if exploreThisRefresh(exploredID, generation, config.ExplorationInterval) && !exploreThisRefresh(omittedID, generation, config.ExplorationInterval) { + break + } + } + if generation == 1000 { + t.Fatal("could not find a deterministic split between exploration buckets") + } + inputs := []SourceLearningCandidate{ + {ID: exploredID, FallbackPriority: 100, Quality: zeroQuality}, + {ID: omittedID, FallbackPriority: 100, Quality: zeroQuality}, + {ID: fmt.Sprintf("%032x", 3), FallbackPriority: 10, Quality: store.SourceQuality{ + AdmissionRefreshes: 3, ZeroAdmissionStreak: 3, LastAdmissionAt: now.Add(-time.Minute), + }}, + {ID: fmt.Sprintf("%032x", 4), FallbackPriority: 100, ProbeContractCurrent: true, Quality: store.SourceQuality{ + AdmissionRefreshes: 3, ZeroAdmissionStreak: 3, LastAdmissionAt: now, + ProbeSamples: 1, ProbePasses: 1, ProbeSuccessfulCompletions: 1, ProbeUpdatedAt: now, + }}, + {ID: fmt.Sprintf("%032x", 5), FallbackPriority: 100, Quality: store.SourceQuality{Corrupt: true, CooldownUntil: now.Add(time.Hour)}}, + } + detailed := BuildSourceLearningPlanDetailed(inputs, generation, now, config) + legacy := BuildSourceLearningPlan(inputs, generation, now, config) + if len(legacy) != len(detailed.Selected) { + t.Fatalf("detailed planner selected %d sources; wrapper selected %d", len(detailed.Selected), len(legacy)) + } + for i := range legacy { + if legacy[i] != detailed.Selected[i] { + t.Fatalf("detailed plan changed ranking at %d: %+v versus %+v", i, detailed.Selected[i], legacy[i]) + } + } + byID := make(map[string]SourceLearningDecision, len(detailed.Decisions)) + for _, decision := range detailed.Decisions { + byID[decision.ID] = decision + } + explored := byID[exploredID] + if !explored.Selected || !explored.Cooled || !explored.AdmissionCooled || !explored.Explored || explored.PlanPosition == nil { + t.Fatalf("explored cooldown decision = %+v", explored) + } + wantReasons := []string{"failure_cooldown", "low_yield_cooldown", "zero_admission_cooldown", "exploration"} + if fmt.Sprint(explored.Reasons) != fmt.Sprint(wantReasons) { + t.Fatalf("explored reasons=%v, want %v", explored.Reasons, wantReasons) + } + omitted := byID[omittedID] + if omitted.Selected || !omitted.Cooled || !omitted.AdmissionCooled || omitted.Explored || omitted.PlanPosition != nil { + t.Fatalf("omitted cooldown decision = %+v", omitted) + } + if fmt.Sprint(omitted.Reasons) != fmt.Sprint(wantReasons[:3]) { + t.Fatalf("omitted reasons=%v, want all overlapping cooldowns %v", omitted.Reasons, wantReasons[:3]) + } + trusted := byID[inputs[2].ID] + if !trusted.Selected || !trusted.TrustedPriorityProtected || trusted.AdmissionCooled { + t.Fatalf("trusted priority source was not protected: %+v", trusted) + } + productive := byID[inputs[3].ID] + if !productive.Selected || !productive.CurrentPositiveProbeProtected || !productive.ProbeContractCurrent || productive.AdmissionCooled { + t.Fatalf("current-contract positive source was not protected: %+v", productive) + } + corrupt := byID[inputs[4].ID] + if !corrupt.Selected || corrupt.Cooled || corrupt.AdmissionCooled || fmt.Sprint(corrupt.Reasons) != "[quality_corrupt_fail_open]" { + t.Fatalf("corrupt quality did not fail open with an explicit reason: %+v", corrupt) + } +} diff --git a/internal/monitor/source_refresh_trace.go b/internal/monitor/source_refresh_trace.go new file mode 100644 index 0000000..67a87f2 --- /dev/null +++ b/internal/monitor/source_refresh_trace.go @@ -0,0 +1,601 @@ +package monitor + +import ( + "bytes" + "compress/gzip" + "context" + "crypto/rand" + "encoding/hex" + "encoding/json" + "errors" + "fmt" + "io" + "time" + + "github.com/Crawlora-org/FreeProxyAPI/internal/store" +) + +const sourceRefreshTraceWriteTimeout = 250 * time.Millisecond +const maxSourceRefreshTraceJSONBytes = 8 << 20 + +type sourceRefreshTrace struct { + RefreshID string `json:"refresh_id"` + Generation uint64 `json:"generation,omitempty"` + CapturedAt *time.Time `json:"captured_at,omitempty"` + StartedAt time.Time `json:"started_at"` + CompletedAt *time.Time `json:"completed_at,omitempty"` + Status string `json:"status"` + Result string `json:"result,omitempty"` + Configured int `json:"configured"` + Selected int `json:"selected"` + Cooled int `json:"cooled"` + Exploring int `json:"exploring"` + AdmissionCooled int `json:"admission_cooled"` + Decisions []sourceRefreshTraceDecision `json:"decisions"` +} + +type sourceRefreshTraceDecision struct { + SourceID string `json:"source_id"` + Selected bool `json:"selected"` + Cooled bool `json:"cooled"` + Explored bool `json:"explored"` + AdmissionCooled bool `json:"admission_cooled"` + TrustedPriorityProtected bool `json:"trusted_priority_protected"` + CurrentPositiveProbeProtected bool `json:"current_positive_probe_protected"` + ProbeContractCurrent bool `json:"probe_contract_current"` + PlanPosition *int `json:"plan_position,omitempty"` + Reasons []string `json:"reasons"` + FetchStatus string `json:"fetch_status"` + MergeStatus string `json:"merge_status"` + AdmissionObservationStatus string `json:"admission_observation_status"` + AdmissionObservationReasons []string `json:"admission_observation_reasons,omitempty"` + FetchAttempts int `json:"fetch_attempts,omitempty"` + HTTPAttempts int64 `json:"http_attempts,omitempty"` + NetworkBytes int64 `json:"network_bytes,omitempty"` + FetchDurationMS int64 `json:"fetch_duration_ms,omitempty"` + Parsed int64 `json:"parsed,omitempty"` + AdmittedCandidates int64 `json:"admitted_candidates,omitempty"` + Truncated bool `json:"truncated,omitempty"` + FetchStartedAt *time.Time `json:"fetch_started_at,omitempty"` + FetchCompletedAt *time.Time `json:"fetch_completed_at,omitempty"` +} + +type sourceRefreshTraceSummary struct { + RefreshID string `json:"refresh_id"` + Generation uint64 `json:"generation,omitempty"` + CapturedAt *time.Time `json:"captured_at,omitempty"` + StartedAt time.Time `json:"started_at"` + CompletedAt *time.Time `json:"completed_at,omitempty"` + Status string `json:"status"` + Result string `json:"result,omitempty"` + Configured int `json:"configured"` + Selected int `json:"selected"` + Cooled int `json:"cooled"` + Exploring int `json:"exploring"` + AdmissionCooled int `json:"admission_cooled"` +} + +type sourceRefreshTraceList struct { + RetentionSeconds int `json:"retention_seconds"` + MaxTraces int `json:"max_traces"` + Traces []sourceRefreshTraceSummary `json:"traces"` +} + +func newSourceRefreshTraceID() (string, error) { + var raw [16]byte + if _, err := rand.Read(raw[:]); err != nil { + return "", err + } + return hex.EncodeToString(raw[:]), nil +} + +func (r *Runner) beginSourceRefreshTrace(ctx context.Context) string { + id, err := newSourceRefreshTraceID() + if err != nil { + r.sourcePlan.mu.Lock() + r.sourcePlan.trace = nil + r.sourcePlan.traceDecisionIndex = nil + r.sourcePlan.mu.Unlock() + r.recordSourceRefreshTraceWriteFailure() + return "" + } + now := r.nowFunc().UTC() + trace := &sourceRefreshTrace{RefreshID: id, StartedAt: now, Status: "started", Decisions: []sourceRefreshTraceDecision{}} + r.sourcePlan.mu.Lock() + r.sourcePlan.trace = trace + r.sourcePlan.traceDecisionIndex = make(map[string]int) + r.sourcePlan.mu.Unlock() + r.persistSourceRefreshTrace(ctx) + return id +} + +func (r *Runner) setSourceRefreshTracePlan(ctx context.Context, generation uint64, capturedAt time.Time, configured int, decisions []SourceLearningDecision) { + r.sourcePlan.mu.Lock() + if r.sourcePlan.trace == nil { + r.sourcePlan.mu.Unlock() + return + } + trace := r.sourcePlan.trace + trace.Generation = generation + captured := capturedAt.UTC() + trace.CapturedAt = &captured + trace.Configured = configured + trace.Decisions = make([]sourceRefreshTraceDecision, 0, len(decisions)) + r.sourcePlan.traceDecisionIndex = make(map[string]int, len(decisions)) + for _, decision := range decisions { + status, mergeStatus := "not_selected", "not_selected" + admissionStatus := "not_selected" + if decision.Selected { + status, mergeStatus = "not_started", "not_started" + admissionStatus = "pending" + if generation == 0 { + admissionStatus = "not_applicable" + } + trace.Selected++ + } + if decision.Cooled { + trace.Cooled++ + } + if decision.Explored { + trace.Exploring++ + } + if decision.AdmissionCooled { + trace.AdmissionCooled++ + } + r.sourcePlan.traceDecisionIndex[decision.ID] = len(trace.Decisions) + trace.Decisions = append(trace.Decisions, sourceRefreshTraceDecision{ + SourceID: decision.ID, Selected: decision.Selected, Cooled: decision.Cooled, + Explored: decision.Explored, AdmissionCooled: decision.AdmissionCooled, + TrustedPriorityProtected: decision.TrustedPriorityProtected, + CurrentPositiveProbeProtected: decision.CurrentPositiveProbeProtected, + ProbeContractCurrent: decision.ProbeContractCurrent, + PlanPosition: decision.PlanPosition, Reasons: append([]string(nil), decision.Reasons...), + FetchStatus: status, MergeStatus: mergeStatus, AdmissionObservationStatus: admissionStatus, + }) + } + r.sourcePlan.mu.Unlock() + r.persistSourceRefreshTrace(ctx) +} + +func (r *Runner) setSourceRefreshTraceGeneration(ctx context.Context, generation uint64, capturedAt time.Time, configured int) { + r.sourcePlan.mu.Lock() + if r.sourcePlan.trace == nil { + r.sourcePlan.mu.Unlock() + return + } + r.sourcePlan.trace.Generation = generation + captured := capturedAt.UTC() + r.sourcePlan.trace.CapturedAt = &captured + r.sourcePlan.trace.Configured = configured + r.sourcePlan.mu.Unlock() + r.persistSourceRefreshTrace(ctx) +} + +func (r *Runner) setLegacySourceRefreshTracePlan(ctx context.Context, specs []SourceSpec) { + decisions := make([]SourceLearningDecision, 0, len(specs)) + seen := make(map[string]struct{}, len(specs)) + for _, spec := range specs { + id := r.sourceID(spec) + if _, exists := seen[id]; exists { + continue + } + seen[id] = struct{}{} + position := len(decisions) + decisions = append(decisions, SourceLearningDecision{ID: id, Selected: true, PlanPosition: &position, Reasons: []string{"learning_disabled"}}) + } + r.setSourceRefreshTracePlan(ctx, 0, r.nowFunc(), len(specs), decisions) +} + +func (r *Runner) recordSourceRefreshFetchStarted(spec SourceSpec) { + id := r.sourceID(spec) + r.sourcePlan.mu.Lock() + if index, exists := r.sourcePlan.traceDecisionIndex[id]; exists && r.sourcePlan.trace != nil && index < len(r.sourcePlan.trace.Decisions) { + decision := &r.sourcePlan.trace.Decisions[index] + if decision.Selected && decision.FetchStatus == "not_started" { + now := r.nowFunc().UTC() + decision.FetchStatus = "started" + decision.FetchStartedAt = &now + } + } + r.sourcePlan.mu.Unlock() +} + +func (r *Runner) recordSourceRefreshFetchResult(ctx context.Context, result sourceRefreshResult) { + id := r.sourceID(result.spec) + r.sourcePlan.mu.Lock() + if index, exists := r.sourcePlan.traceDecisionIndex[id]; exists && r.sourcePlan.trace != nil && index < len(r.sourcePlan.trace.Decisions) { + decision := &r.sourcePlan.trace.Decisions[index] + if decision.Selected { + now := r.nowFunc().UTC() + decision.FetchCompletedAt = &now + decision.FetchAttempts = result.fetchAttempts + decision.HTTPAttempts = result.httpAttempts + decision.NetworkBytes = result.networkBytes + decision.FetchDurationMS = result.fetchDurationMS + decision.Parsed = int64(result.parsed) + decision.Truncated = result.truncated + switch { + case result.stage == sourceStageFetch && result.err != nil && (ctx.Err() != nil || errors.Is(result.err, context.Canceled) || errors.Is(result.err, context.DeadlineExceeded)): + decision.FetchStatus = "cancelled" + case result.stage == sourceStageFetch && result.err != nil: + decision.FetchStatus = "failed" + case result.truncated: + decision.FetchStatus = "truncated" + default: + decision.FetchStatus = "fetched" + } + switch { + case result.stage == sourceStageMerge && result.err != nil: + decision.MergeStatus = "failed" + case decision.FetchStatus == "cancelled" || decision.FetchStatus == "failed": + decision.MergeStatus = "not_reached" + default: + decision.MergeStatus = "completed" + } + if decision.AdmissionObservationStatus != "not_applicable" { + if decision.FetchStatus == "cancelled" { + setAdmissionObservationOutcome(decision, "cancelled", "cancelled") + } else { + censored := false + if decision.FetchStatus == "failed" { + setAdmissionObservationOutcome(decision, "censored", "fetch_failed") + censored = true + } + if decision.Truncated { + setAdmissionObservationOutcome(decision, "censored", "truncated") + censored = true + } + if decision.MergeStatus == "failed" { + setAdmissionObservationOutcome(decision, "censored", "merge_failed") + censored = true + } + if !censored && decision.MergeStatus == "completed" { + decision.AdmissionObservationStatus = "pending" + } + } + } + } + } + r.sourcePlan.mu.Unlock() +} + +func (r *Runner) recordSourceRefreshAdmissionDelta(sourceID string, added int64) { + if added <= 0 { + return + } + r.sourcePlan.mu.Lock() + if index, exists := r.sourcePlan.traceDecisionIndex[sourceID]; exists && r.sourcePlan.trace != nil && index < len(r.sourcePlan.trace.Decisions) { + r.sourcePlan.trace.Decisions[index].AdmittedCandidates += added + } + r.sourcePlan.mu.Unlock() +} + +func (r *Runner) recordSourceRefreshAdmissionCensored(sourceID, reason string) { + r.sourcePlan.mu.Lock() + if index, exists := r.sourcePlan.traceDecisionIndex[sourceID]; exists && r.sourcePlan.trace != nil && index < len(r.sourcePlan.trace.Decisions) { + decision := &r.sourcePlan.trace.Decisions[index] + if decision.Selected && decision.AdmissionObservationStatus != "not_applicable" { + setAdmissionObservationOutcome(decision, "censored", reason) + } + } + r.sourcePlan.mu.Unlock() +} + +func (r *Runner) recordSourceRefreshAdmissionRecorded(sourceIDs []string, failed bool) { + r.sourcePlan.mu.Lock() + defer r.sourcePlan.mu.Unlock() + for _, sourceID := range sourceIDs { + if index, exists := r.sourcePlan.traceDecisionIndex[sourceID]; exists && r.sourcePlan.trace != nil && index < len(r.sourcePlan.trace.Decisions) { + decision := &r.sourcePlan.trace.Decisions[index] + if failed { + setAdmissionObservationOutcome(decision, "write_failed", "observation_write_failed") + } else if decision.AdmissionObservationStatus == "pending" { + decision.AdmissionObservationStatus = "recorded" + } + } + } +} + +func setAdmissionObservationOutcome(decision *sourceRefreshTraceDecision, status, reason string) { + decision.AdmissionObservationStatus = status + for _, existing := range decision.AdmissionObservationReasons { + if existing == reason { + return + } + } + decision.AdmissionObservationReasons = append(decision.AdmissionObservationReasons, reason) +} + +func (r *Runner) finishSourceRefreshTrace(ctx context.Context, originalCtx context.Context, err error) { + r.sourcePlan.mu.Lock() + if r.sourcePlan.trace == nil { + r.sourcePlan.mu.Unlock() + return + } + if err != nil { + if originalCtx.Err() != nil || errors.Is(err, context.Canceled) || errors.Is(err, context.DeadlineExceeded) { + r.sourcePlan.trace.Status = "cancelled" + r.sourcePlan.trace.Result = "cancelled" + for i := range r.sourcePlan.trace.Decisions { + decision := &r.sourcePlan.trace.Decisions[i] + if decision.Selected { + if decision.FetchStatus == "started" || decision.FetchStatus == "not_started" { + decision.FetchStatus = "cancelled" + decision.MergeStatus = "not_reached" + } + if decision.AdmissionObservationStatus == "pending" { + setAdmissionObservationOutcome(decision, "cancelled", "cancelled") + } + } + } + } else { + r.sourcePlan.trace.Status = "failed" + r.sourcePlan.trace.Result = "error" + for i := range r.sourcePlan.trace.Decisions { + decision := &r.sourcePlan.trace.Decisions[i] + if decision.Selected { + if decision.FetchStatus == "started" || decision.FetchStatus == "not_started" { + decision.FetchStatus = "failed" + decision.MergeStatus = "not_reached" + setAdmissionObservationOutcome(decision, "censored", "refresh_failed_before_fetch") + } else if decision.AdmissionObservationStatus == "pending" { + setAdmissionObservationOutcome(decision, "censored", "refresh_failed") + } + } + } + } + } else { + r.sourcePlan.trace.Status = "completed" + if r.sourcePlan.trace.Selected == 0 { + r.sourcePlan.trace.Result = "skipped" + } else { + r.sourcePlan.trace.Result = "ok" + for _, decision := range r.sourcePlan.trace.Decisions { + if decision.Selected && (decision.FetchStatus != "fetched" || (decision.AdmissionObservationStatus != "recorded" && decision.AdmissionObservationStatus != "not_applicable")) { + r.sourcePlan.trace.Result = "partial" + break + } + } + } + } + completed := r.nowFunc().UTC() + r.sourcePlan.trace.CompletedAt = &completed + r.sourcePlan.mu.Unlock() + r.persistSourceRefreshTrace(ctx) +} + +func (r *Runner) persistSourceRefreshTrace(ctx context.Context) { + trace := r.sourceRefreshTraceSnapshot() + if trace.RefreshID == "" || r.store == nil { + return + } + payload, err := encodeSourceRefreshTrace(trace) + if err != nil { + r.recordSourceRefreshTraceWriteFailure() + return + } + summary, err := json.Marshal(summarizeSourceRefreshTrace(trace)) + if err != nil { + r.recordSourceRefreshTraceWriteFailure() + return + } + writeCtx, cancel := context.WithTimeout(context.WithoutCancel(ctx), sourceRefreshTraceWriteTimeout) + defer cancel() + if err := r.store.SaveSourceRefreshTrace(writeCtx, trace.RefreshID, payload, summary, trace.StartedAt); err != nil { + r.recordSourceRefreshTraceWriteFailure() + } +} + +func (r *Runner) sourceRefreshTraceSnapshot() sourceRefreshTrace { + r.sourcePlan.mu.RLock() + defer r.sourcePlan.mu.RUnlock() + if r.sourcePlan.trace == nil { + return sourceRefreshTrace{} + } + trace := *r.sourcePlan.trace + trace.Decisions = make([]sourceRefreshTraceDecision, len(r.sourcePlan.trace.Decisions)) + copy(trace.Decisions, r.sourcePlan.trace.Decisions) + for i := range trace.Decisions { + trace.Decisions[i].Reasons = append([]string(nil), trace.Decisions[i].Reasons...) + trace.Decisions[i].AdmissionObservationReasons = append([]string(nil), trace.Decisions[i].AdmissionObservationReasons...) + } + return trace +} + +func (r *Runner) recordSourceRefreshTraceWriteFailure() { + if r.metrics != nil { + r.metrics.AddSourceRefreshTraceWriteFailure() + } +} + +func summarizeSourceRefreshTrace(trace sourceRefreshTrace) sourceRefreshTraceSummary { + return sourceRefreshTraceSummary{ + RefreshID: trace.RefreshID, Generation: trace.Generation, CapturedAt: trace.CapturedAt, + StartedAt: trace.StartedAt, CompletedAt: trace.CompletedAt, Status: trace.Status, + Result: trace.Result, Configured: trace.Configured, Selected: trace.Selected, + Cooled: trace.Cooled, Exploring: trace.Exploring, AdmissionCooled: trace.AdmissionCooled, + } +} + +func decodeSourceRefreshTraceSummary(payload []byte, expectedID string) (sourceRefreshTraceSummary, error) { + var summary sourceRefreshTraceSummary + if err := json.Unmarshal(payload, &summary); err != nil { + return sourceRefreshTraceSummary{}, errors.New("invalid source refresh trace summary") + } + if !validSourceRefreshTraceID(expectedID) || summary.RefreshID != expectedID || !validRefreshTraceStatus(summary.Status) || !validRefreshTraceResult(summary.Result) || summary.Configured < 0 || summary.Configured > store.MaxLearningSources || summary.Selected < 0 || summary.Selected > summary.Configured || summary.Cooled < 0 || summary.Cooled > summary.Configured || summary.Exploring < 0 || summary.Exploring > summary.Selected || summary.AdmissionCooled < 0 || summary.AdmissionCooled > summary.Configured { + return sourceRefreshTraceSummary{}, errors.New("invalid source refresh trace summary") + } + if summary.Status == "started" { + if summary.CompletedAt != nil || summary.Result != "" { + return sourceRefreshTraceSummary{}, errors.New("invalid source refresh trace summary lifecycle") + } + } else if summary.CompletedAt == nil || summary.Result == "" { + return sourceRefreshTraceSummary{}, errors.New("invalid source refresh trace summary lifecycle") + } + return summary, nil +} + +func decodeSourceRefreshTrace(payload []byte) (sourceRefreshTrace, error) { + if len(payload) == 0 { + return sourceRefreshTrace{}, errors.New("invalid source refresh trace") + } + if len(payload) >= 2 && payload[0] == 0x1f && payload[1] == 0x8b { + reader, err := gzip.NewReader(bytes.NewReader(payload)) + if err != nil { + return sourceRefreshTrace{}, errors.New("invalid source refresh trace") + } + decompressed, readErr := io.ReadAll(io.LimitReader(reader, maxSourceRefreshTraceJSONBytes+1)) + closeErr := reader.Close() + if readErr != nil || closeErr != nil || len(decompressed) > maxSourceRefreshTraceJSONBytes { + return sourceRefreshTrace{}, errors.New("invalid source refresh trace") + } + payload = decompressed + } else if len(payload) > maxSourceRefreshTraceJSONBytes { + return sourceRefreshTrace{}, errors.New("invalid source refresh trace") + } + var trace sourceRefreshTrace + if err := json.Unmarshal(payload, &trace); err != nil { + return sourceRefreshTrace{}, fmt.Errorf("decode source refresh trace") + } + if !validSourceRefreshTraceID(trace.RefreshID) || !validRefreshTraceStatus(trace.Status) || !validRefreshTraceResult(trace.Result) || trace.Decisions == nil || len(trace.Decisions) > store.MaxLearningSources || trace.Configured < len(trace.Decisions) || trace.Configured > store.MaxLearningSources || trace.Selected < 0 || trace.Selected > trace.Configured { + return sourceRefreshTrace{}, errors.New("invalid source refresh trace") + } + if trace.Status == "started" { + if trace.CompletedAt != nil || trace.Result != "" { + return sourceRefreshTrace{}, errors.New("invalid source refresh trace lifecycle") + } + } else if trace.CompletedAt == nil || trace.Result == "" { + return sourceRefreshTrace{}, errors.New("invalid source refresh trace lifecycle") + } + selected, cooled, exploring, admissionCooled := 0, 0, 0, 0 + positions := make(map[int]struct{}, trace.Selected) + for _, decision := range trace.Decisions { + if !validSourceRefreshTraceID(decision.SourceID) || !validRefreshFetchStatus(decision.FetchStatus) || !validRefreshMergeStatus(decision.MergeStatus) || !validAdmissionObservationStatus(decision.AdmissionObservationStatus) || len(decision.Reasons) > 8 || len(decision.AdmissionObservationReasons) > 8 { + return sourceRefreshTrace{}, errors.New("invalid source refresh trace decision") + } + if decision.Selected { + selected++ + if decision.FetchStatus == "not_selected" || decision.MergeStatus == "not_selected" || decision.AdmissionObservationStatus == "not_selected" || decision.PlanPosition == nil || *decision.PlanPosition < 0 || *decision.PlanPosition >= trace.Selected { + return sourceRefreshTrace{}, errors.New("invalid selected source trace decision") + } + if _, exists := positions[*decision.PlanPosition]; exists { + return sourceRefreshTrace{}, errors.New("duplicate source refresh plan position") + } + positions[*decision.PlanPosition] = struct{}{} + } else if decision.FetchStatus != "not_selected" || decision.MergeStatus != "not_selected" || decision.AdmissionObservationStatus != "not_selected" || decision.PlanPosition != nil || decision.Explored { + return sourceRefreshTrace{}, errors.New("invalid omitted source trace decision") + } + if decision.Explored && (!decision.Selected || !decision.Cooled) { + return sourceRefreshTrace{}, errors.New("invalid exploration trace decision") + } + if decision.Cooled { + cooled++ + } + if decision.Explored { + if !decision.Cooled { + return sourceRefreshTrace{}, errors.New("invalid exploration trace decision") + } + exploring++ + } + if decision.AdmissionCooled { + admissionCooled++ + } + for _, reason := range decision.Reasons { + switch reason { + case "failure_cooldown", "low_yield_cooldown", "zero_admission_cooldown", "exploration", "learning_disabled", "quality_corrupt_fail_open", "active": + default: + return sourceRefreshTrace{}, errors.New("invalid source refresh trace reason") + } + } + if decision.AdmissionCooled && !containsTraceReason(decision.Reasons, "zero_admission_cooldown") { + return sourceRefreshTrace{}, errors.New("missing zero-admission trace reason") + } + if decision.Cooled && !containsTraceReason(decision.Reasons, "failure_cooldown") && !containsTraceReason(decision.Reasons, "low_yield_cooldown") && !containsTraceReason(decision.Reasons, "zero_admission_cooldown") { + return sourceRefreshTrace{}, errors.New("missing source cooldown trace reason") + } + for _, reason := range decision.AdmissionObservationReasons { + switch reason { + case "truncated", "candidate_cap", "fetch_failed", "merge_failed", "cancelled", "observation_write_failed", "refresh_failed_before_fetch", "refresh_failed": + default: + return sourceRefreshTrace{}, errors.New("invalid source refresh trace admission reason") + } + } + } + if selected != trace.Selected || cooled != trace.Cooled || exploring != trace.Exploring || admissionCooled != trace.AdmissionCooled { + return sourceRefreshTrace{}, errors.New("invalid source refresh trace counts") + } + return trace, nil +} + +func containsTraceReason(reasons []string, want string) bool { + for _, reason := range reasons { + if reason == want { + return true + } + } + return false +} + +func encodeSourceRefreshTrace(trace sourceRefreshTrace) ([]byte, error) { + encoded, err := json.Marshal(trace) + if err != nil { + return nil, fmt.Errorf("encode source refresh trace") + } + if len(encoded) > maxSourceRefreshTraceJSONBytes { + return nil, errors.New("source refresh trace exceeds size limit") + } + var compressed bytes.Buffer + writer := gzip.NewWriter(&compressed) + if _, err := writer.Write(encoded); err != nil { + return nil, fmt.Errorf("compress source refresh trace") + } + if err := writer.Close(); err != nil { + return nil, fmt.Errorf("compress source refresh trace") + } + return compressed.Bytes(), nil +} + +func validRefreshTraceStatus(status string) bool { + switch status { + case "started", "completed", "failed", "cancelled": + return true + default: + return false + } +} + +func validRefreshTraceResult(result string) bool { + switch result { + case "", "ok", "partial", "skipped", "error", "cancelled": + return true + default: + return false + } +} + +func validRefreshFetchStatus(status string) bool { + switch status { + case "not_selected", "not_started", "started", "fetched", "truncated", "failed", "cancelled": + return true + default: + return false + } +} + +func validRefreshMergeStatus(status string) bool { + switch status { + case "not_selected", "not_started", "not_reached", "completed", "failed": + return true + default: + return false + } +} + +func validAdmissionObservationStatus(status string) bool { + switch status { + case "not_selected", "not_applicable", "pending", "recorded", "censored", "write_failed", "cancelled": + return true + default: + return false + } +} diff --git a/internal/monitor/source_refresh_trace_test.go b/internal/monitor/source_refresh_trace_test.go new file mode 100644 index 0000000..1a367f4 --- /dev/null +++ b/internal/monitor/source_refresh_trace_test.go @@ -0,0 +1,334 @@ +package monitor + +import ( + "context" + "encoding/json" + "fmt" + "net/http" + "net/http/httptest" + "path/filepath" + "strings" + "testing" + "time" + + "github.com/Crawlora-org/FreeProxyAPI/internal/store" +) + +func TestRefreshTraceLinksExplorationDecisionToFetchedSource(t *testing.T) { + dir := t.TempDir() + base := time.Date(2026, 10, 13, 12, 0, 0, 0, time.UTC) + var cooled SourceSpec + for i := 0; ; i++ { + candidate := syntheticSource(t, dir, fmt.Sprintf("cooled-%d.txt", i), "192.0.2.201:8080\n") + id := StableSourceID(candidate.URL, candidate.defaultScheme("http")) + if exploreThisRefresh(id, 1, 10) { + cooled = candidate + break + } + } + active := syntheticSource(t, dir, "active.txt", "192.0.2.202:8080\n") + runner, _ := newSourceLearningRunner(t, "trace-lifecycle:v1", []SourceSpec{cooled, active}, true) + runner.config.SourceLearningZeroAdmissionCooldown = time.Hour + runner.config.SourceLearningZeroAdmissionRefreshes = 3 + ctx := context.Background() + cooledID := runner.sourceID(cooled) + for i := 1; i <= 3; i++ { + if err := runner.store.RecordSourceAdmissionRefresh(ctx, map[string]int64{cooledID: 0}, base.Add(time.Duration(i-4)*time.Minute)); err != nil { + t.Fatal(err) + } + } + if err := runner.store.RecordSourceFetchFailure(ctx, cooledID, base); err != nil { + t.Fatal(err) + } + if err := runner.refresh(ctx); err != nil { + t.Fatalf("refresh: %v", err) + } + trace := runner.sourceRefreshTraceSnapshot() + if len(trace.RefreshID) != 32 || trace.Generation != 1 || trace.Status != "completed" || trace.Result != "ok" || trace.CompletedAt == nil { + t.Fatalf("refresh lifecycle trace = %+v", trace) + } + var decision *sourceRefreshTraceDecision + for i := range trace.Decisions { + if trace.Decisions[i].SourceID == cooledID { + decision = &trace.Decisions[i] + break + } + } + if decision == nil { + t.Fatal("exploratory source missing from exact decision set") + } + if !decision.Selected || !decision.Cooled || !decision.AdmissionCooled || !decision.Explored || decision.FetchStatus != "fetched" || decision.FetchAttempts != 1 || decision.HTTPAttempts != 0 || decision.AdmissionObservationStatus != "recorded" { + t.Fatalf("exploratory fetch decision = %+v", *decision) + } + if strings.Join(decision.Reasons, ",") != "failure_cooldown,zero_admission_cooldown,exploration" { + t.Fatalf("overlapping reasons = %v", decision.Reasons) + } + quality, err := runner.store.LoadSourceQuality(ctx, []string{cooledID}, base) + if err != nil || quality[cooledID].LastAdmitted == 0 || quality[cooledID].ZeroAdmissionStreak != 0 { + t.Fatalf("exploration did not import/reset the source after planning: quality=%+v err=%v", quality[cooledID], err) + } +} + +func TestRefreshTraceRecordsFailedCancelledAndCensoredOutcomes(t *testing.T) { + t.Run("failed", func(t *testing.T) { + dir := t.TempDir() + bad := SourceSpec{URL: "file://" + dir + "/missing.txt", Scheme: "http"} + runner, _ := newSourceLearningRunner(t, "trace-failed:v1", []SourceSpec{bad}, true) + if err := runner.refresh(context.Background()); err == nil { + t.Fatal("refresh with only a failed feed succeeded") + } + trace := runner.sourceRefreshTraceSnapshot() + if trace.Status != "failed" || trace.Result != "error" || trace.Generation != 1 || len(trace.Decisions) != 1 || trace.Decisions[0].FetchStatus != "failed" || trace.Decisions[0].MergeStatus != "not_reached" || trace.Decisions[0].AdmissionObservationStatus != "censored" || strings.Join(trace.Decisions[0].AdmissionObservationReasons, ",") != "fetch_failed" { + t.Fatalf("failed refresh trace = %+v", trace) + } + }) + + t.Run("partial_failure", func(t *testing.T) { + dir := t.TempDir() + good := syntheticSource(t, dir, "good.txt", "192.0.2.210:8080\n") + bad := SourceSpec{URL: "file://" + filepath.Join(dir, "missing-partial.txt"), Scheme: "http"} + runner, _ := newSourceLearningRunner(t, "trace-partial-failure:v1", []SourceSpec{good, bad}, true) + if err := runner.refresh(context.Background()); err != nil { + t.Fatalf("one failed feed should not abort other source imports: %v", err) + } + trace := runner.sourceRefreshTraceSnapshot() + if trace.Status != "completed" || trace.Result != "partial" || len(trace.Decisions) != 2 { + t.Fatalf("partial source failure trace = %+v", trace) + } + statuses := make(map[string]string, len(trace.Decisions)) + for _, decision := range trace.Decisions { + statuses[decision.SourceID] = decision.FetchStatus + } + if statuses[runner.sourceID(good)] != "fetched" || statuses[runner.sourceID(bad)] != "failed" { + t.Fatalf("per-source fetch outcomes = %v", statuses) + } + }) + + t.Run("cancelled", func(t *testing.T) { + spec := SourceSpec{URL: "file:///synthetic/cancelled-feed", Scheme: "http"} + runner, _ := newSourceLearningRunner(t, "trace-cancelled:v1", []SourceSpec{spec}, true) + ctx, cancel := context.WithCancel(context.Background()) + calls := 0 + runner.nowFunc = func() time.Time { + calls++ + if calls == 3 { // begin, planner snapshot, then first selected fetch start + cancel() + } + return time.Date(2026, 10, 13, 12, 0, 0, 0, time.UTC) + } + if err := runner.refresh(ctx); err == nil { + t.Fatal("cancelled refresh returned nil") + } + trace := runner.sourceRefreshTraceSnapshot() + if trace.Status != "cancelled" || trace.Result != "cancelled" || len(trace.Decisions) != 1 || trace.Decisions[0].FetchStatus != "cancelled" || trace.Decisions[0].AdmissionObservationStatus != "cancelled" { + t.Fatalf("cancelled refresh trace = %+v", trace) + } + }) + + t.Run("truncated", func(t *testing.T) { + dir := t.TempDir() + var filename string + for i := 0; i < 1000; i++ { + candidateName := fmt.Sprintf("truncated-%d.txt", i) + candidateURL := "file://" + filepath.Join(dir, candidateName) + if exploreThisRefresh(StableSourceID(candidateURL, "http"), 1, 10) { + filename = candidateName + break + } + } + if filename == "" { + t.Fatal("could not find a source ID in the generation-1 exploration bucket") + } + spec := syntheticSource(t, dir, filename, feedLines("trace-truncated", 12)) + runner, _ := newSourceLearningRunner(t, "trace-truncated:v1", []SourceSpec{spec}, true) + runner.config.SourceLearningZeroAdmissionCooldown = time.Hour + runner.config.SourceLearningZeroAdmissionRefreshes = 3 + runner.config.MaxRecordsPerSource = 2 + ctx := context.Background() + id := runner.sourceID(spec) + base := runner.nowFunc() + for i := 1; i <= 3; i++ { + if err := runner.store.RecordSourceAdmissionRefresh(ctx, map[string]int64{id: 0}, base.Add(time.Duration(i-4)*time.Minute)); err != nil { + t.Fatal(err) + } + } + if err := runner.refresh(context.Background()); err != nil { + t.Fatalf("truncated refresh: %v", err) + } + trace := runner.sourceRefreshTraceSnapshot() + if trace.Status != "completed" || trace.Result != "partial" || len(trace.Decisions) != 1 || !trace.Decisions[0].Explored || !trace.Decisions[0].AdmissionCooled || trace.Decisions[0].FetchStatus != "truncated" || !trace.Decisions[0].Truncated || trace.Decisions[0].AdmittedCandidates != 2 || trace.Decisions[0].AdmissionObservationStatus != "censored" || strings.Join(trace.Decisions[0].AdmissionObservationReasons, ",") != "truncated" { + t.Fatalf("truncated refresh trace = %+v", trace) + } + quality, err := runner.store.LoadSourceQuality(ctx, []string{id}, base) + if err != nil || quality[id].Admitted != 2 || quality[id].AdmissionRefreshes != 3 || quality[id].ZeroAdmissionStreak != 0 || quality[id].LastAdmitted != 0 || !quality[id].LastAdmissionAt.Equal(base.Add(-time.Minute)) { + t.Fatalf("partial positive truncation changed full-observation history incorrectly: quality=%+v err=%v", quality[id], err) + } + }) + + t.Run("candidate_cap", func(t *testing.T) { + runner, mini, _ := capRunner(t, `,"max_candidates":1`, feedLines("trace-candidate-cap", 12)) + runner.config.SourceLearningEnabled = true + runner.config.MaxCandidates = 0 + if err := runner.refresh(context.Background()); err != nil { + t.Fatalf("initial refresh: %v", err) + } + runner.config.MaxCandidates = 1 + mini.FastForward(time.Hour) + if err := runner.refresh(context.Background()); err != nil { + t.Fatalf("candidate-capped refresh: %v", err) + } + trace := runner.sourceRefreshTraceSnapshot() + if trace.Status != "completed" || trace.Result != "partial" || len(trace.Decisions) != 1 || trace.Decisions[0].FetchStatus != "fetched" || trace.Decisions[0].AdmissionObservationStatus != "censored" || strings.Join(trace.Decisions[0].AdmissionObservationReasons, ",") != "candidate_cap" { + t.Fatalf("candidate-capped refresh trace = %+v", trace) + } + id := runner.sourceID(runner.config.SourceSpecs[0]) + quality, err := runner.store.LoadSourceQuality(context.Background(), []string{id}, time.Now()) + if err != nil || quality[id].AdmissionRefreshes != 1 { + t.Fatalf("candidate-capped source recorded a full admission observation: quality=%+v err=%v", quality[id], err) + } + }) + +} + +func TestTraceWriteFailureDoesNotInterruptImports(t *testing.T) { + dir := t.TempDir() + spec := syntheticSource(t, dir, "trace-fail-open.txt", "192.0.2.203:8080\n") + runner, mini := newSourceLearningRunner(t, "trace-write-fail:v1", []SourceSpec{spec}, true) + if err := mini.Set("trace-write-fail:v1:source-refresh-traces", "wrong-type"); err != nil { + t.Fatal(err) + } + if err := runner.refresh(context.Background()); err != nil { + t.Fatalf("telemetry write failure interrupted refresh: %v", err) + } + stats, err := runner.store.Stats(context.Background()) + if err != nil || stats.Candidates != 1 { + t.Fatalf("candidate import after trace failure: stats=%+v err=%v", stats, err) + } + metrics := httptest.NewRecorder() + runner.metrics.Handler().ServeHTTP(metrics, httptest.NewRequest(http.MethodGet, "/metrics", nil)) + foundFailureCount := false + for _, line := range strings.Split(metrics.Body.String(), "\n") { + fields := strings.Fields(line) + if len(fields) == 2 && fields[0] == "freeproxyapi_source_refresh_trace_write_failures_total" && fields[1] != "0" { + foundFailureCount = true + break + } + } + if !foundFailureCount { + t.Fatal("trace write failure metric was not exposed") + } +} + +func TestSourceRefreshTraceEncodingBoundsMaximumCatalog(t *testing.T) { + decisions := make([]sourceRefreshTraceDecision, store.MaxLearningSources) + selectedCount, cooled, exploring, admissionCooled := 0, 0, 0, 0 + for i := range decisions { + isCooled := i%3 == 0 + isExplored := isCooled && i%30 == 0 + selected := !isCooled || isExplored + fetchStatus, mergeStatus, admissionStatus := "not_selected", "not_selected", "not_selected" + var position *int + if selected { + positionValue := selectedCount + position = &positionValue + selectedCount++ + fetchStatus, mergeStatus, admissionStatus = "fetched", "completed", "recorded" + } + reasons := []string{"active"} + if isCooled { + cooled++ + admissionCooled++ + reasons = []string{"zero_admission_cooldown"} + } + if isExplored { + exploring++ + reasons = append(reasons, "exploration") + } + decisions[i] = sourceRefreshTraceDecision{ + SourceID: fmt.Sprintf("%032x", i+1), Selected: selected, Cooled: isCooled, + Explored: isExplored, AdmissionCooled: isCooled, + TrustedPriorityProtected: i%100 == 0, ProbeContractCurrent: true, + PlanPosition: position, Reasons: reasons, FetchStatus: fetchStatus, + MergeStatus: mergeStatus, AdmissionObservationStatus: admissionStatus, + } + } + trace := sourceRefreshTrace{ + RefreshID: "aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa", Generation: 1, StartedAt: time.Now().UTC(), + Status: "completed", Result: "ok", Configured: len(decisions), Selected: selectedCount, Cooled: cooled, Exploring: exploring, AdmissionCooled: admissionCooled, + Decisions: decisions, + } + completed := time.Now().UTC() + trace.CompletedAt = &completed + payload, err := encodeSourceRefreshTrace(trace) + if err != nil { + t.Fatalf("encode maximum catalog trace: %v", err) + } + if len(payload) > 1<<20 { + t.Fatalf("compressed maximum catalog trace is %d bytes, exceeds Redis record limit", len(payload)) + } + decoded, err := decodeSourceRefreshTrace(payload) + if err != nil || len(decoded.Decisions) != store.MaxLearningSources { + t.Fatalf("decode maximum catalog trace returned %d decisions, err=%v", len(decoded.Decisions), err) + } +} + +func TestSourceRefreshTraceRouteIsPrivateAndRejectsCorruption(t *testing.T) { + runner, _, _ := newSourceEfficiencyTestRunner(t, []SourceSpec{{URL: "file:///synthetic/trace-route", Scheme: "http"}}) + runner.config.AdminListenAddr = "127.0.0.1:0" + const traceID = "aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa" + now := time.Now().UTC() + summary, err := json.Marshal(sourceRefreshTraceSummary{RefreshID: traceID, StartedAt: now, Status: "started"}) + if err != nil { + t.Fatal(err) + } + if err := runner.store.SaveSourceRefreshTrace(context.Background(), traceID, []byte("not-json"), summary, now); err != nil { + t.Fatal(err) + } + server, err := startHealthServer("127.0.0.1:0", runner) + if err != nil { + t.Fatal(err) + } + defer server.Shutdown() + public := httptest.NewRecorder() + server.server.Handler.ServeHTTP(public, httptest.NewRequest(http.MethodGet, "/report/source-refreshes?refresh_id="+traceID, nil)) + if public.Code != http.StatusNotFound { + t.Fatalf("public trace route status=%d, want 404", public.Code) + } + list := httptest.NewRecorder() + server.admin.Handler.ServeHTTP(list, httptest.NewRequest(http.MethodGet, "/report/source-refreshes", nil)) + if list.Code != http.StatusOK || strings.Contains(list.Body.String(), "not-json") || strings.Contains(list.Body.String(), "decisions") { + t.Fatalf("trace summary list status=%d body=%s", list.Code, list.Body.String()) + } + admin := httptest.NewRecorder() + server.admin.Handler.ServeHTTP(admin, httptest.NewRequest(http.MethodGet, "/report/source-refreshes?refresh_id="+traceID, nil)) + if admin.Code != http.StatusServiceUnavailable || strings.Contains(admin.Body.String(), "not-json") || strings.Contains(admin.Body.String(), "source_id") { + t.Fatalf("corrupt trace response status=%d body=%s", admin.Code, admin.Body.String()) + } + if strings.Contains(admin.Body.String(), "http") || strings.Contains(admin.Body.String(), "proxy") { + t.Fatalf("corrupt trace failure leaked payload data: %s", admin.Body.String()) + } +} + +func TestSourceRefreshTraceRouteRequiresLoopbackWithoutAdminListener(t *testing.T) { + runner, _, _ := newSourceEfficiencyTestRunner(t, []SourceSpec{{URL: "file:///synthetic/trace-loopback", Scheme: "http"}}) + runner.config.AdminListenAddr = "" + server, err := startHealthServer("127.0.0.1:0", runner) + if err != nil { + t.Fatal(err) + } + defer server.Shutdown() + remote := httptest.NewRequest(http.MethodGet, "/report/source-refreshes", nil) + remote.RemoteAddr = "203.0.113.12:30000" + blocked := httptest.NewRecorder() + server.server.Handler.ServeHTTP(blocked, remote) + if blocked.Code != http.StatusNotFound { + t.Fatalf("non-loopback trace route status=%d, want 404", blocked.Code) + } + local := httptest.NewRequest(http.MethodGet, "/report/source-refreshes", nil) + local.RemoteAddr = "127.0.0.1:30001" + allowed := httptest.NewRecorder() + server.server.Handler.ServeHTTP(allowed, local) + if allowed.Code != http.StatusOK { + t.Fatalf("loopback trace route status=%d body=%s, want 200", allowed.Code, allowed.Body.String()) + } +} diff --git a/internal/store/source_refresh_trace.go b/internal/store/source_refresh_trace.go new file mode 100644 index 0000000..c6a0fdd --- /dev/null +++ b/internal/store/source_refresh_trace.go @@ -0,0 +1,159 @@ +package store + +import ( + "context" + "errors" + "fmt" + "strconv" + "time" + + "github.com/redis/go-redis/v9" +) + +const ( + MaxSourceRefreshTraces = 64 + SourceRefreshTraceRetention = 72 * time.Hour + maxSourceRefreshTraceBytes = 1 << 20 + maxSourceRefreshTraceList = MaxSourceRefreshTraces +) + +var ErrSourceRefreshTraceNotFound = errors.New("source refresh trace not found") + +func (s *Redis) sourceRefreshTraceIndexKey() string { + return s.prefix + ":source-refresh-traces" +} + +func (s *Redis) sourceRefreshTraceKey(id string) string { + return s.prefix + ":source-refresh-trace:" + id +} + +func (s *Redis) sourceRefreshTraceSummaryKey(id string) string { + return s.prefix + ":source-refresh-trace-summary:" + id +} + +var saveSourceRefreshTraceScript = redis.NewScript(` +local index = KEYS[1] +local payload = KEYS[2] +local summary = KEYS[3] +local id = ARGV[1] +local data = ARGV[2] +local captured = tonumber(ARGV[3]) +local ttl = tonumber(ARGV[4]) +local max = tonumber(ARGV[5]) +local index_type = redis.call('TYPE', index).ok +if index_type ~= 'none' and index_type ~= 'zset' then + return redis.error_reply('WRONGTYPE source refresh trace index is not a sorted set') +end +redis.call('SET', payload, data, 'PX', ttl) +redis.call('SET', summary, ARGV[8], 'PX', ttl) +redis.call('ZADD', index, captured, id) +local excess = redis.call('ZCARD', index) - max +if excess > 0 then + local evicted = redis.call('ZRANGE', index, 0, excess - 1) + for _, old_id in ipairs(evicted) do + redis.call('DEL', ARGV[6] .. old_id) + redis.call('DEL', ARGV[7] .. old_id) + end + redis.call('ZREMRANGEBYRANK', index, 0, excess - 1) +end +redis.call('PEXPIRE', index, ttl) +return 1 +`) + +// SaveSourceRefreshTrace stores an opaque bounded detail payload and compact +// summary, both indexed by capture time. The oldest keys are deleted when the +// index evicts their ID. +func (s *Redis) SaveSourceRefreshTrace(ctx context.Context, id string, payload, summary []byte, capturedAt time.Time) error { + if !validSourceID(id) { + return errors.New("invalid source refresh trace ID") + } + if len(payload) == 0 || len(payload) > maxSourceRefreshTraceBytes { + return fmt.Errorf("source refresh trace payload size must be between 1 and %d bytes", maxSourceRefreshTraceBytes) + } + if len(summary) == 0 || len(summary) > 16<<10 { + return errors.New("source refresh trace summary size must be between 1 and 16384 bytes") + } + if capturedAt.IsZero() || capturedAt.UnixMilli() <= 0 { + return errors.New("invalid source refresh trace capture time") + } + _, err := saveSourceRefreshTraceScript.Run(ctx, s.client, + []string{s.sourceRefreshTraceIndexKey(), s.sourceRefreshTraceKey(id), s.sourceRefreshTraceSummaryKey(id)}, + id, payload, capturedAt.UnixMilli(), SourceRefreshTraceRetention.Milliseconds(), + MaxSourceRefreshTraces, s.prefix+":source-refresh-trace:", s.prefix+":source-refresh-trace-summary:", summary).Result() + if err != nil { + return fmt.Errorf("save source refresh trace: %w", err) + } + return nil +} + +// LoadSourceRefreshTraceSummary returns the compact summary for a listed ID. +func (s *Redis) LoadSourceRefreshTraceSummary(ctx context.Context, id string) ([]byte, error) { + if !validSourceID(id) { + return nil, errors.New("invalid source refresh trace ID") + } + _, err := s.client.ZScore(ctx, s.sourceRefreshTraceIndexKey(), id).Result() + if err != nil { + if errors.Is(err, redis.Nil) { + return nil, ErrSourceRefreshTraceNotFound + } + return nil, fmt.Errorf("check source refresh trace index: %w", err) + } + summary, err := s.client.Get(ctx, s.sourceRefreshTraceSummaryKey(id)).Bytes() + if err != nil { + if errors.Is(err, redis.Nil) { + return nil, ErrSourceRefreshTraceNotFound + } + return nil, fmt.Errorf("load source refresh trace summary: %w", err) + } + return summary, nil +} + +// LoadSourceRefreshTrace returns the opaque detail bytes for an indexed trace. +func (s *Redis) LoadSourceRefreshTrace(ctx context.Context, id string) ([]byte, error) { + if !validSourceID(id) { + return nil, errors.New("invalid source refresh trace ID") + } + _, err := s.client.ZScore(ctx, s.sourceRefreshTraceIndexKey(), id).Result() + if err != nil { + if errors.Is(err, redis.Nil) { + return nil, ErrSourceRefreshTraceNotFound + } + return nil, fmt.Errorf("check source refresh trace index: %w", err) + } + payload, err := s.client.Get(ctx, s.sourceRefreshTraceKey(id)).Bytes() + if err != nil { + if errors.Is(err, redis.Nil) { + return nil, ErrSourceRefreshTraceNotFound + } + return nil, fmt.Errorf("load source refresh trace: %w", err) + } + return payload, nil +} + +// ListSourceRefreshTraceIDs returns the newest indexed IDs, newest first. +func (s *Redis) ListSourceRefreshTraceIDs(ctx context.Context, limit int) ([]string, error) { + if limit < 0 || limit > maxSourceRefreshTraceList { + return nil, fmt.Errorf("source refresh trace limit must be between 0 and %d", maxSourceRefreshTraceList) + } + if limit == 0 { + return []string{}, nil + } + // Read at most the index's configured bound even if external corruption has + // inserted arbitrary members, then expose only validated source IDs. + ids, err := s.client.ZRangeArgs(ctx, redis.ZRangeArgs{ + Key: s.sourceRefreshTraceIndexKey(), Start: "0", Stop: strconv.Itoa(MaxSourceRefreshTraces - 1), Rev: true, + }).Result() + if err != nil { + return nil, fmt.Errorf("list source refresh trace IDs: %w", err) + } + valid := make([]string, 0, limit) + for _, id := range ids { + if validSourceID(id) { + valid = append(valid, id) + if len(valid) == limit { + break + } + } + } + return valid, nil +} diff --git a/internal/store/source_refresh_trace_test.go b/internal/store/source_refresh_trace_test.go new file mode 100644 index 0000000..e6b7b20 --- /dev/null +++ b/internal/store/source_refresh_trace_test.go @@ -0,0 +1,147 @@ +package store + +import ( + "context" + "errors" + "fmt" + "strings" + "testing" + "time" + + "github.com/alicebob/miniredis/v2" +) + +func newSourceRefreshTraceTestStore(t *testing.T) (*miniredis.Miniredis, *Redis) { + t.Helper() + mini := miniredis.RunT(t) + s, err := NewRedis("redis://"+mini.Addr()+"/0", "trace-test:v1") + if err != nil { + t.Fatalf("NewRedis: %v", err) + } + t.Cleanup(func() { _ = s.Close() }) + return mini, s +} + +func traceTestID(n int) string { return fmt.Sprintf("%032x", n) } + +func TestSourceRefreshTraceValidationAndLimits(t *testing.T) { + _, s := newSourceRefreshTraceTestStore(t) + ctx := context.Background() + now := time.Date(2026, 10, 14, 0, 0, 0, 0, time.UTC) + for _, id := range []string{"", "ABCDEF0123456789ABCDEF0123456789", "0123456789abcdef0123456789abcdeg", "0123456789abcdef0123456789abcde"} { + if err := s.SaveSourceRefreshTrace(ctx, id, []byte(`{}`), []byte(`{}`), now); err == nil { + t.Fatalf("SaveSourceRefreshTrace accepted invalid ID %q", id) + } + if _, err := s.LoadSourceRefreshTrace(ctx, id); err == nil { + t.Fatalf("LoadSourceRefreshTrace accepted invalid ID %q", id) + } + } + if err := s.SaveSourceRefreshTrace(ctx, traceTestID(1), nil, []byte(`{}`), now); err == nil { + t.Fatal("SaveSourceRefreshTrace accepted an empty payload") + } + if err := s.SaveSourceRefreshTrace(ctx, traceTestID(1), []byte(`{}`), []byte(`{}`), time.Time{}); err == nil { + t.Fatal("SaveSourceRefreshTrace accepted a zero capture time") + } + if err := s.SaveSourceRefreshTrace(ctx, traceTestID(1), []byte(strings.Repeat("x", maxSourceRefreshTraceBytes+1)), []byte(`{}`), now); err == nil { + t.Fatal("SaveSourceRefreshTrace accepted an oversized payload") + } + for _, limit := range []int{-1, MaxSourceRefreshTraces + 1} { + if _, err := s.ListSourceRefreshTraceIDs(ctx, limit); err == nil { + t.Fatalf("ListSourceRefreshTraceIDs accepted limit %d", limit) + } + } +} + +func TestSourceRefreshTraceRetentionAndPayloadEviction(t *testing.T) { + mini, s := newSourceRefreshTraceTestStore(t) + ctx := context.Background() + base := time.Date(2026, 10, 14, 0, 0, 0, 0, time.UTC) + for i := 1; i <= MaxSourceRefreshTraces+1; i++ { + if err := s.SaveSourceRefreshTrace(ctx, traceTestID(i), []byte(fmt.Sprintf(`{"n":%d}`, i)), []byte(`{"summary":true}`), base.Add(time.Duration(i)*time.Second)); err != nil { + t.Fatalf("SaveSourceRefreshTrace(%d): %v", i, err) + } + } + ids, err := s.ListSourceRefreshTraceIDs(ctx, MaxSourceRefreshTraces) + if err != nil { + t.Fatalf("ListSourceRefreshTraceIDs: %v", err) + } + if len(ids) != MaxSourceRefreshTraces || ids[0] != traceTestID(MaxSourceRefreshTraces+1) || ids[len(ids)-1] != traceTestID(2) { + t.Fatalf("retained IDs = %d entries, first %q last %q", len(ids), ids[0], ids[len(ids)-1]) + } + if mini.Exists(s.sourceRefreshTraceKey(traceTestID(1))) { + t.Fatal("evicted payload key still exists") + } + if mini.Exists(s.sourceRefreshTraceSummaryKey(traceTestID(1))) { + t.Fatal("evicted summary key still exists") + } + if _, err := s.LoadSourceRefreshTrace(ctx, traceTestID(1)); !errors.Is(err, ErrSourceRefreshTraceNotFound) { + t.Fatalf("LoadSourceRefreshTrace(evicted) error = %v", err) + } + if ttl := mini.TTL(s.sourceRefreshTraceKey(traceTestID(2))); ttl <= 0 || ttl > SourceRefreshTraceRetention { + t.Fatalf("payload TTL = %s", ttl) + } + if ttl := mini.TTL(s.sourceRefreshTraceSummaryKey(traceTestID(2))); ttl <= 0 || ttl > SourceRefreshTraceRetention { + t.Fatalf("summary TTL = %s", ttl) + } + if ttl := mini.TTL(s.sourceRefreshTraceIndexKey()); ttl <= 0 || ttl > SourceRefreshTraceRetention { + t.Fatalf("index TTL = %s", ttl) + } + mini.FastForward(SourceRefreshTraceRetention) + if mini.Exists(s.sourceRefreshTraceIndexKey()) || mini.Exists(s.sourceRefreshTraceKey(traceTestID(MaxSourceRefreshTraces+1))) || mini.Exists(s.sourceRefreshTraceSummaryKey(traceTestID(MaxSourceRefreshTraces+1))) { + t.Fatal("trace index, payload, or summary survived retention") + } +} + +func TestSourceRefreshTraceMissingAndCorruptRedisTypes(t *testing.T) { + mini, s := newSourceRefreshTraceTestStore(t) + ctx := context.Background() + id := traceTestID(7) + if _, err := s.LoadSourceRefreshTrace(ctx, id); !errors.Is(err, ErrSourceRefreshTraceNotFound) { + t.Fatalf("missing trace error = %v", err) + } + if err := s.SaveSourceRefreshTrace(ctx, id, []byte(`{"opaque":true}`), []byte(`{"summary":true}`), time.Now().UTC()); err != nil { + t.Fatalf("SaveSourceRefreshTrace: %v", err) + } + got, err := s.LoadSourceRefreshTrace(ctx, id) + if err != nil || string(got) != `{"opaque":true}` { + t.Fatalf("LoadSourceRefreshTrace = %q, %v", got, err) + } + gotSummary, err := s.LoadSourceRefreshTraceSummary(ctx, id) + if err != nil || string(gotSummary) != `{"summary":true}` { + t.Fatalf("LoadSourceRefreshTraceSummary = %q, %v", gotSummary, err) + } + mini.Del(s.sourceRefreshTraceKey(id)) + if _, err := s.LoadSourceRefreshTrace(ctx, id); !errors.Is(err, ErrSourceRefreshTraceNotFound) { + t.Fatalf("trace with missing payload error = %v", err) + } + mini.Set(s.sourceRefreshTraceKey(id), `{"opaque":true}`) + mini.ZAdd(s.sourceRefreshTraceIndexKey(), float64(time.Now().Add(time.Hour).UnixMilli()), "https://invalid.example/feed") + listed, err := s.ListSourceRefreshTraceIDs(ctx, MaxSourceRefreshTraces) + if err != nil { + t.Fatalf("ListSourceRefreshTraceIDs after invalid member: %v", err) + } + for _, listedID := range listed { + if !validSourceID(listedID) { + t.Fatalf("ListSourceRefreshTraceIDs exposed an invalid member %q", listedID) + } + } + mini.Del(s.sourceRefreshTraceIndexKey()) + mini.Set(s.sourceRefreshTraceIndexKey(), "wrong-type") + if _, err := s.ListSourceRefreshTraceIDs(ctx, 1); err == nil { + t.Fatal("ListSourceRefreshTraceIDs hid a wrong-type index") + } + if err := s.SaveSourceRefreshTrace(ctx, traceTestID(8), []byte(`{}`), []byte(`{}`), time.Now().UTC()); err == nil { + t.Fatal("SaveSourceRefreshTrace hid a wrong-type index") + } + + // Restore an index entry and poison its payload key to cover GET wrong-type. + mini.Del(s.sourceRefreshTraceIndexKey()) + if err := s.SaveSourceRefreshTrace(ctx, id, []byte(`{}`), []byte(`{}`), time.Now().UTC()); err != nil { + t.Fatalf("restore index: %v", err) + } + mini.Del(s.sourceRefreshTraceKey(id)) + mini.HSet(s.sourceRefreshTraceKey(id), "field", "wrong-type") + if _, err := s.LoadSourceRefreshTrace(ctx, id); err == nil { + t.Fatal("LoadSourceRefreshTrace hid a wrong-type payload") + } +}