diff --git a/docs/configuration.md b/docs/configuration.md index 1b1800c..6304072 100644 --- a/docs/configuration.md +++ b/docs/configuration.md @@ -91,7 +91,11 @@ toward this guard; failed, skipped, truncated, or candidate-cap-affected passes are excluded. 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. +in effect for cooled feeds. Any actual new candidate insertion clears that +source's zero-admission streak immediately, including an insertion made before +a later partial, truncated, or capped pass ends. This immediate reset does not +count as a completed observation or update the last full-refresh admission +fields; duplicate or tombstone-blocked records do not reset the streak. The setting is off by default and has no environment override. Quality hashes expire after 30 days, and only the active configured catalog is attributed. diff --git a/docs/redis-schema.md b/docs/redis-schema.md index b55b22c..960b195 100644 --- a/docs/redis-schema.md +++ b/docs/redis-schema.md @@ -50,13 +50,20 @@ The same hash may also hold bounded measurement counters: - `admitted` accumulates candidate IDs newly inserted into the tracked candidate set by atomic upsert, not rows staged by a feed. It can include real insertions - from a refresh that later fails. -- `admission_refreshes`, `last_admitted`, `zero_admission_streak`, and - `last_admission_ms` describe completed whole-refresh observations. They are + from a refresh that later fails. Any such positive insertion immediately + resets `zero_admission_streak`, even if the containing pass later fails or is + truncated or capped. +- `admission_refreshes`, `last_admitted`, and `last_admission_ms` describe + completed whole-refresh observations. They are updated only when a source participated in a successful, uncapped and untruncated refresh pass. Failed, skipped, truncated, or candidate-cap-affected source passes are excluded. `zero_admission_streak` counts consecutive - qualifying observations with zero new candidate IDs. + qualifying observations with zero new candidate IDs since the latest + positive admission. A partial positive + insertion resets that streak without incrementing `admission_refreshes` or + changing `last_admitted`/`last_admission_ms`; those fields still describe only + completed full-refresh observations. Duplicate and tombstone-blocked upserts + do not reset the streak. - `http_attempts`, `network_bytes`, `fetch_duration_ms`, and `fetch_cost_observations` accumulate source-fetch measurements. Attempts count HTTP client calls; bytes count consumed HTTP response-body reader bytes, diff --git a/internal/monitor/source_efficiency_integration_test.go b/internal/monitor/source_efficiency_integration_test.go index 70336be..d98978f 100644 --- a/internal/monitor/source_efficiency_integration_test.go +++ b/internal/monitor/source_efficiency_integration_test.go @@ -73,6 +73,48 @@ func TestSourceAdmissionCooldownUsesStoredFullRefreshObservations(t *testing.T) } } +func TestPartialPositiveAdmissionClearsCooldownWithoutFullRefreshObservation(t *testing.T) { + ctx := context.Background() + now := time.Date(2026, 10, 14, 12, 0, 0, 0, time.UTC) + runner, _, clock := newSourceEfficiencyTestRunner(t, []SourceSpec{{URL: "file:///synthetic/partial-admission", Scheme: "http"}}) + *clock = now + runner.config.SourceLearningZeroAdmissionCooldown = time.Hour + runner.config.SourceLearningZeroAdmissionRefreshes = 3 + id := runner.sourceID(runner.config.SourceSpecs[0]) + if err := runner.store.SetActiveLearningSources(ctx, []string{id}); err != nil { + t.Fatal(err) + } + for i := 1; i <= 3; i++ { + if err := runner.store.RecordSourceAdmissionRefresh(ctx, map[string]int64{id: 0}, now.Add(time.Duration(i)*time.Minute)); err != nil { + t.Fatal(err) + } + } + + quality, err := runner.store.LoadSourceQuality(ctx, []string{id}, now.Add(3*time.Minute)) + if err != nil || quality[id].ZeroAdmissionStreak != 3 || quality[id].AdmissionRefreshes != 3 { + t.Fatalf("pre-upsert quality=%+v err=%v", quality[id], err) + } + // This direct source-aware upsert models a partial/capped pass: it commits + // one genuine SADD but deliberately records no completed-refresh observation. + counts, err := runner.store.UpsertPrioritizedSourceCounts(ctx, []string{"http://partial-new.example:8080"}, now.Add(4*time.Minute), id) + if err != nil || counts.Added != 1 { + t.Fatalf("partial admission=%+v err=%v", counts, err) + } + quality, err = runner.store.LoadSourceQuality(ctx, []string{id}, now.Add(4*time.Minute)) + if err != nil { + t.Fatal(err) + } + if quality[id].Admitted != 1 || quality[id].ZeroAdmissionStreak != 0 || quality[id].AdmissionRefreshes != 3 || quality[id].LastAdmitted != 0 || !quality[id].LastAdmissionAt.Equal(now.Add(3*time.Minute)) { + t.Fatalf("partial admission changed full-observation fields: %+v", quality[id]) + } + + config := SourceLearningConfig{Enabled: true, MinSamples: 8, ZeroAdmissionCooldown: time.Hour, ZeroAdmissionRefreshes: 3, ExplorationInterval: 10} + plan := BuildSourceLearningPlan([]SourceLearningCandidate{{ID: id, FallbackPriority: 20, Quality: quality[id]}}, 1, now.Add(4*time.Minute), config) + if len(plan) != 1 || plan[0].AdmissionCooled { + t.Fatalf("source with a partial positive admission remained cooled: %+v", plan) + } +} + func TestCappedRefreshDoesNotRecordAdmissionObservation(t *testing.T) { runner, mini, _ := capRunner(t, `, "max_candidates": 1`, feedLines("cap", 12)) runner.config.MaxCandidates = 0 diff --git a/internal/store/redis.go b/internal/store/redis.go index ce3c03c..0d763f4 100644 --- a/internal/store/redis.go +++ b/internal/store/redis.go @@ -401,14 +401,16 @@ local last_pair = #ARGV local source_id = nil if #ARGV % 2 == 0 then source_id = ARGV[#ARGV]; last_pair = #ARGV - 1 end local quality_key = nil +local quality_hash_safe = false local record_admission = false if source_id then quality_key = KEYS[7] - -- Efficiency telemetry must never prevent candidate imports. Only enable - -- its increment when the existing hash field is a safe integer with room - -- for every possible insertion in this bounded batch. + -- Efficiency telemetry must never prevent candidate imports. A safe hash + -- can have its zero-admission streak reset after a real insertion even when + -- its cumulative admitted counter is malformed or full. local current_raw = redis.pcall('HGET', quality_key, 'admitted') if not (type(current_raw) == 'table' and current_raw.err) then + quality_hash_safe = true if current_raw == false then current_raw = '0' end local current = tonumber(current_raw) local max_batch = math.floor((last_pair - 1) / 2) @@ -455,8 +457,14 @@ for i = 2, last_pair, 2 do end end end -if source_id and record_admission and source_admitted > 0 then - redis.call('HINCRBY', quality_key, 'admitted', tostring(source_admitted)) +if source_id and quality_hash_safe and source_admitted > 0 then + -- A partial or capped refresh may still have admitted genuinely new IDs. + -- Clear stale zero-yield history immediately, but leave full-refresh + -- observation fields to RecordSourceAdmissionRefresh. + redis.call('HSET', quality_key, 'zero_admission_streak', '0') + if record_admission then + redis.call('HINCRBY', quality_key, 'admitted', tostring(source_admitted)) + end redis.call('PEXPIRE', quality_key, '2592000000') end return {added, skipped} diff --git a/internal/store/source_efficiency_test.go b/internal/store/source_efficiency_test.go index f6820b5..2aedf4f 100644 --- a/internal/store/source_efficiency_test.go +++ b/internal/store/source_efficiency_test.go @@ -118,6 +118,69 @@ func TestSourceAdmissionCountsOnlyActualInsertionsAndTracksCompletedRefreshes(t } } +func TestPartialPositiveUpsertResetsStreakWithoutRecordingFullRefresh(t *testing.T) { + ctx := context.Background() + _, s := newTestStore(t, "source-admission-partial-positive:v1") + now := time.Unix(1_800_000_000, 0) + id := StableSourceID("https://feeds.example/partial-positive", "http") + if err := s.SetActiveLearningSources(ctx, []string{id}); err != nil { + t.Fatal(err) + } + for i := 1; i <= 3; i++ { + if err := s.RecordSourceAdmissionRefresh(ctx, map[string]int64{id: 0}, now.Add(time.Duration(i)*time.Minute)); err != nil { + t.Fatal(err) + } + } + quality, err := s.LoadSourceQuality(ctx, []string{id}, now.Add(3*time.Minute)) + if err != nil || quality[id].ZeroAdmissionStreak != 3 || quality[id].AdmissionRefreshes != 3 { + t.Fatalf("pre-upsert admission state=%+v err=%v", quality[id], err) + } + + counts, err := s.UpsertPrioritizedSourceCounts(ctx, []string{"http://new-source.example:8080"}, now.Add(4*time.Minute), id) + if err != nil || counts.Added != 1 { + t.Fatalf("partial positive upsert=%+v err=%v", counts, err) + } + quality, err = s.LoadSourceQuality(ctx, []string{id}, now.Add(4*time.Minute)) + if err != nil { + t.Fatal(err) + } + got := quality[id] + if got.Admitted != 1 || got.ZeroAdmissionStreak != 0 || got.AdmissionRefreshes != 3 || got.LastAdmitted != 0 || !got.LastAdmissionAt.Equal(now.Add(3*time.Minute)) { + t.Fatalf("partial upsert should reset only zero history, got %+v", got) + } +} + +func TestDuplicateAndTombstoneOnlyUpsertsDoNotResetZeroAdmissionStreak(t *testing.T) { + ctx := context.Background() + mini, s := newTestStore(t, "source-admission-nonpositive-upsert:v1") + now := time.Unix(1_800_000_000, 0) + id := StableSourceID("https://feeds.example/nonpositive", "http") + if err := s.SetActiveLearningSources(ctx, []string{id}); err != nil { + t.Fatal(err) + } + existing := "http://already-present.example:8080" + if _, err := s.UpsertPrioritizedSourceCounts(ctx, []string{existing}, now, id); err != nil { + t.Fatal(err) + } + for i := 1; i <= 3; i++ { + if err := s.RecordSourceAdmissionRefresh(ctx, map[string]int64{id: 0}, now.Add(time.Duration(i)*time.Minute)); err != nil { + t.Fatal(err) + } + } + blocked := "http://tombstoned.example:8080" + if err := mini.Set(s.tombstoneKey(proxyID(blocked)), fmt.Sprintf("1:%d", now.Add(time.Hour).UnixMilli())); err != nil { + t.Fatal(err) + } + counts, err := s.UpsertPrioritizedSourceCounts(ctx, []string{existing, blocked}, now.Add(4*time.Minute), id) + if err != nil || counts.Added != 0 || counts.Tombstoned != 1 { + t.Fatalf("duplicate/tombstoned upsert=%+v err=%v", counts, err) + } + quality, err := s.LoadSourceQuality(ctx, []string{id}, now.Add(4*time.Minute)) + if err != nil || quality[id].ZeroAdmissionStreak != 3 || quality[id].AdmissionRefreshes != 3 || quality[id].LastAdmitted != 0 || !quality[id].LastAdmissionAt.Equal(now.Add(3*time.Minute)) { + t.Fatalf("nonpositive upsert changed completed history: quality=%+v err=%v", quality[id], err) + } +} + func TestCorruptAdmissionTelemetryNeverBlocksCandidateImports(t *testing.T) { ctx := context.Background() mini, s := newTestStore(t, "source-admission-corrupt:v1") @@ -131,16 +194,19 @@ func TestCorruptAdmissionTelemetryNeverBlocksCandidateImports(t *testing.T) { name string seed func(key string) wantStored string + wantStreak string }{ { name: "malformed counter", - seed: func(key string) { mini.HSet(key, "admitted", "not-an-integer") }, + seed: func(key string) { mini.HSet(key, "admitted", "not-an-integer", "zero_admission_streak", "3") }, wantStored: "not-an-integer", + wantStreak: "0", }, { name: "counter at exact cap", - seed: func(key string) { mini.HSet(key, "admitted", fmt.Sprint(int64(1<<53-1))) }, + seed: func(key string) { mini.HSet(key, "admitted", fmt.Sprint(int64(1<<53-1)), "zero_admission_streak", "3") }, wantStored: fmt.Sprint(int64(1<<53 - 1)), + wantStreak: "0", }, { name: "wrong-type quality key", @@ -150,6 +216,7 @@ func TestCorruptAdmissionTelemetryNeverBlocksCandidateImports(t *testing.T) { } }, wantStored: "legacy non-hash value", + wantStreak: "", }, } for index, tc := range tests { @@ -182,6 +249,11 @@ func TestCorruptAdmissionTelemetryNeverBlocksCandidateImports(t *testing.T) { if err != nil || stored != tc.wantStored { t.Fatalf("corrupt admission counter changed: got=%q err=%v want=%q", stored, err, tc.wantStored) } + if tc.name != "wrong-type quality key" { + if got := mini.HGet(idKey, "zero_admission_streak"); got != tc.wantStreak { + t.Fatalf("zero admission streak=%q, want reset to %q", got, tc.wantStreak) + } + } }) } }