Merge pull request 'Two maps grow for the process lifetime with no eviction' (#194) from task/641-two-maps-grow-for-the-process-lifetime-w into master

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
+	seen    map[string]*seenIDs // feed → item IDs, for items with no date
-	polled  map[string]bool            // feed → polled at least once in THIS process
+	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
 }
 

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

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
 		}

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