From c8eb494aeef294f8ced78832f98eb758f4f41c22 Mon Sep 17 00:00:00 2001 From: tonywangcn Date: Wed, 14 Oct 2026 15:32:16 +0800 Subject: [PATCH 1/2] Trace source refresh decisions Record the source planner decisions and fetch outcomes from the refresh that uses them. Retain bounded, opaque traces on the operator route so cooldown eligibility, exploration, censored observations, and actual fetches remain distinguishable without storing source or proxy URLs. --- docs/api.md | 49 +- docs/configuration.md | 16 +- docs/redis-schema.md | 24 + internal/monitor/health.go | 97 +++ internal/monitor/metrics.go | 58 +- internal/monitor/runner.go | 49 +- internal/monitor/source_learning.go | 123 +++- internal/monitor/source_learning_runner.go | 27 +- internal/monitor/source_learning_test.go | 77 +++ internal/monitor/source_refresh_trace.go | 601 ++++++++++++++++++ internal/monitor/source_refresh_trace_test.go | 334 ++++++++++ internal/store/source_refresh_trace.go | 156 +++++ internal/store/source_refresh_trace_test.go | 147 +++++ 13 files changed, 1686 insertions(+), 72 deletions(-) create mode 100644 internal/monitor/source_refresh_trace.go create mode 100644 internal/monitor/source_refresh_trace_test.go create mode 100644 internal/store/source_refresh_trace.go create mode 100644 internal/store/source_refresh_trace_test.go 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..a4f4891 --- /dev/null +++ b/internal/store/source_refresh_trace.go @@ -0,0 +1,156 @@ +package store + +import ( + "context" + "errors" + "fmt" + "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.ZRevRange(ctx, s.sourceRefreshTraceIndexKey(), 0, MaxSourceRefreshTraces-1).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") + } +} From c608da98978bf27e09f41c3cb7e8b926806600eb Mon Sep 17 00:00:00 2001 From: tonywangcn Date: Wed, 14 Oct 2026 15:37:34 +0800 Subject: [PATCH 2/2] Use current Redis range query Avoid the deprecated ZRevRange call while preserving the bounded newest-first trace index read. --- internal/store/source_refresh_trace.go | 5 ++++- 1 file changed, 4 insertions(+), 1 deletion(-) diff --git a/internal/store/source_refresh_trace.go b/internal/store/source_refresh_trace.go index a4f4891..c6a0fdd 100644 --- a/internal/store/source_refresh_trace.go +++ b/internal/store/source_refresh_trace.go @@ -4,6 +4,7 @@ import ( "context" "errors" "fmt" + "strconv" "time" "github.com/redis/go-redis/v9" @@ -139,7 +140,9 @@ func (s *Redis) ListSourceRefreshTraceIDs(ctx context.Context, limit int) ([]str } // 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.ZRevRange(ctx, s.sourceRefreshTraceIndexKey(), 0, MaxSourceRefreshTraces-1).Result() + 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) }