From f7b76c572f0a130e5379c4ee8ff1af3972d95266 Mon Sep 17 00:00:00 2001 From: claude Date: Fri, 7 Aug 2026 01:32:15 +0400 Subject: [PATCH 1/2] Bound the undated-item set per feed (V-641) rss.Poller.seen held every undated item ever seen, one entry per id, for as long as mavend ran. fresh() added and nothing removed. A feed that ships items with no grew it forever. seenIDs is the same set with a bound: the map answers the lookup, a slice remembers insertion order, and the oldest id falls out past 512. The cap has to stay above any one feed's front page or an item still listed there would be written a second time, and a few hundred covers the largest page anyone publishes. The set only ever had to span one poll window plus the resync guard, not all of history. Dedupe behaviour is unchanged. The comment at fresh() explains why the set does not survive a restart; it never bounded it within one run. --- internal/rss/poller.go | 43 +++++++++++++++++++++++++++++++------ internal/rss/poller_test.go | 25 +++++++++++++++++++++ 2 files changed, 62 insertions(+), 6 deletions(-) diff --git a/internal/rss/poller.go b/internal/rss/poller.go index 353619c..e6d5c52 100644 --- a/internal/rss/poller.go +++ b/internal/rss/poller.go @@ -89,8 +89,8 @@ type Poller struct { ranker Ranker cfg Config nextDue map[string]time.Time - seen map[string]map[string]bool // feed → item ID, for items with no date - polled map[string]bool // feed → polled at least once in THIS process + seen map[string]*seenIDs // feed → item IDs, for items with no date + polled map[string]bool // feed → polled at least once in THIS process } // NewPoller wires a poller. Returns nil when there is nothing to poll — a @@ -122,7 +122,7 @@ func NewPoller(feeds []FeedConfig, fetch Fetcher, notes Notes, marks Marks, embe feeds: valid, fetch: fetch, notes: notes, marks: marks, embed: embed, ranker: ranker, cfg: cfg, nextDue: map[string]time.Time{}, - seen: map[string]map[string]bool{}, + seen: map[string]*seenIDs{}, polled: map[string]bool{}, } } @@ -283,6 +283,38 @@ func (p *Poller) mark(ctx context.Context, feed string, now time.Time) (time.Tim return at, true } +// maxSeenPerFeed bounds the undated-item set. It has to stay comfortably above +// any one feed's front page, or an item still listed there would fall out of the +// set and be written a second time. A few hundred entries covers the largest +// page anyone publishes, and the set only has to span one poll window plus the +// resync guard, not all of history. +const maxSeenPerFeed = 512 + +// seenIDs is a bounded insertion-ordered set. The map answers the lookup, the +// slice remembers what to drop first, so an undated feed cannot grow the poller +// for as long as mavend runs. +type seenIDs struct { + ids map[string]bool + order []string +} + +// add records id and reports whether it was new. +func (s *seenIDs) add(id string) bool { + if s.ids == nil { + s.ids = make(map[string]bool, maxSeenPerFeed) + } + if s.ids[id] { + return false + } + s.ids[id] = true + s.order = append(s.order, id) + if len(s.order) > maxSeenPerFeed { + delete(s.ids, s.order[0]) + s.order = s.order[1:] + } + return true +} + // fresh — two dedup rules, because feeds are inconsistent about dates. A dated // item must be newer than the mark; an undated one is kept once per process by // ID. @@ -308,12 +340,11 @@ func (p *Poller) fresh(f FeedConfig, it Item, mark, now time.Time, resync bool) id = it.Title } if p.seen[f.Name] == nil { - p.seen[f.Name] = map[string]bool{} + p.seen[f.Name] = &seenIDs{} } - if p.seen[f.Name][id] { + if !p.seen[f.Name].add(id) { return false } - p.seen[f.Name][id] = true return !resync } diff --git a/internal/rss/poller_test.go b/internal/rss/poller_test.go index f071399..28ed267 100644 --- a/internal/rss/poller_test.go +++ b/internal/rss/poller_test.go @@ -3,6 +3,7 @@ package rss import ( "context" "errors" + "fmt" "strings" "testing" "time" @@ -208,3 +209,27 @@ func TestNoFeedsMeansNoPoller(t *testing.T) { t.Fatal("a feed with no name or url is not a configuration") } } + +// An undated feed used to grow p.seen for as long as mavend ran. The set is +// bounded now, and the bound must not cost the dedupe an item still on the +// front page — only ids far older than any page fall out. +func TestSeenIDsBounded(t *testing.T) { + var s seenIDs + for i := 0; i < maxSeenPerFeed*3; i++ { + if !s.add(fmt.Sprintf("item-%d", i)) { + t.Fatalf("item-%d read as already seen", i) + } + if len(s.ids) > maxSeenPerFeed || len(s.order) > maxSeenPerFeed { + t.Fatalf("after %d inserts: ids=%d order=%d, cap is %d", + i+1, len(s.ids), len(s.order), maxSeenPerFeed) + } + } + // The newest insert is still deduped; the oldest was evicted. + last := fmt.Sprintf("item-%d", maxSeenPerFeed*3-1) + if s.add(last) { + t.Fatalf("%s read as new, so the most recent id was dropped", last) + } + if !s.add("item-0") { + t.Fatal("item-0 survived, so nothing was evicted") + } +} From 4be6852b94fcada2c278da78e794a72f88e1f34d Mon Sep 17 00:00:00 2001 From: claude Date: Fri, 7 Aug 2026 01:32:26 +0400 Subject: [PATCH 2/2] Drop host rate-limit entries that can no longer delay anything (V-641) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit webfetch.Fetcher.last held one entry per distinct host the crawler ever dialed, never pruned. Bounded in practice by how many hosts get crawled, but crawl.on_demand is true in deploy, so the host set is whatever he names out loud. An entry older than HostInterval cannot delay a request — waitTurn would let the next one straight through — so it is dropped. The sweep runs on write and only once the map passes 64 entries, below which walking it costs more than the entries do. Rate limiting is unchanged: a host dialed inside the interval is kept, which the test asserts, because pruning one would hand out a free turn. --- internal/webfetch/webfetch.go | 22 +++++++++++++++++++ internal/webfetch/webfetch_test.go | 35 ++++++++++++++++++++++++++++++ 2 files changed, 57 insertions(+) diff --git a/internal/webfetch/webfetch.go b/internal/webfetch/webfetch.go index aea5284..1ebedde 100644 --- a/internal/webfetch/webfetch.go +++ b/internal/webfetch/webfetch.go @@ -295,6 +295,27 @@ func (f *Fetcher) checkURL(u *url.URL) error { return nil } +// pruneHostsAbove is when pruneLocked bothers to walk the map. Below it the +// walk costs more than the entries do, and `crawl.on_demand` means the host set +// is whatever he names out loud, so it grows slowly. +const pruneHostsAbove = 64 + +// pruneLocked drops hosts whose last dial is further back than HostInterval. +// Such an entry cannot delay anything — waitTurn would let the next request +// through immediately — so keeping it only holds memory for the life of the +// process. Caller holds f.mu. +func (f *Fetcher) pruneLocked(now time.Time) { + if len(f.last) <= pruneHostsAbove { + return + } + cutoff := now.Add(-f.cfg.HostInterval) + for h, at := range f.last { + if at.Before(cutoff) { + delete(f.last, h) + } + } +} + // waitTurn blocks until this host's rate-limit interval has elapsed. It holds // no lock while sleeping, so two hosts never wait on each other. func (f *Fetcher) waitTurn(ctx context.Context, host string) error { @@ -304,6 +325,7 @@ func (f *Fetcher) waitTurn(ctx context.Context, host string) error { earliest := f.last[host].Add(f.cfg.HostInterval) if !now.Before(earliest) { f.last[host] = now + f.pruneLocked(now) f.mu.Unlock() return nil } diff --git a/internal/webfetch/webfetch_test.go b/internal/webfetch/webfetch_test.go index 3eb8d3a..2ec2df2 100644 --- a/internal/webfetch/webfetch_test.go +++ b/internal/webfetch/webfetch_test.go @@ -4,6 +4,7 @@ import ( "bytes" "context" "errors" + "fmt" "io" "net" "net/http" @@ -318,3 +319,37 @@ func TestPostObeysDenylist(t *testing.T) { t.Fatalf("error = %v, want ErrBlocked", err) } } + +// f.last used to hold one entry per host ever dialed, for the life of the +// process. A host whose last dial is older than HostInterval cannot delay +// anything, so it is dropped once the map is worth walking. +func TestHostRateMapIsPruned(t *testing.T) { + f := New(Config{HostInterval: time.Minute, AllowPrivate: true}) + stale := time.Now().Add(-time.Hour) + for i := 0; i < pruneHostsAbove*2; i++ { + f.last[fmt.Sprintf("h%d.example", i)] = stale + } + + // One real turn is what triggers the sweep. + if err := f.waitTurn(context.Background(), "fresh.example"); err != nil { + t.Fatal(err) + } + if len(f.last) != 1 { + t.Fatalf("len(f.last) = %d after the sweep, want 1 (only the host just dialed)", len(f.last)) + } + if _, ok := f.last["fresh.example"]; !ok { + t.Fatal("the host just dialed was pruned, so its own rate limit is lost") + } + + // A host inside the interval is kept: pruning must not hand out a free turn. + f.last["recent.example"] = time.Now() + for i := 0; i < pruneHostsAbove*2; i++ { + f.last[fmt.Sprintf("g%d.example", i)] = stale + } + if err := f.waitTurn(context.Background(), "other.example"); err != nil { + t.Fatal(err) + } + if _, ok := f.last["recent.example"]; !ok { + t.Fatal("a host dialed inside HostInterval was pruned") + } +}