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") + } +} 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") + } +}