Keep prefixes that do not parse out of the prefix tables (closes #114)

UpdatePrefixesBatch now skips a prefix that does not parse, so it never
becomes a row in prefixes_v4 or prefixes_v6 and never counts in the prefix
total in the stats. Before, the prefix handler stored no route for such a
prefix and logged it once per prefix and peer, but still passed it to
UpdatePrefixesBatch, which stored it.

Model: opus-5-5

internal/database/database.go

@@ -855,7 +855,7 @@ func insertPrefixesInTable(
 
 // UpdatePrefixesBatch stores, in a single transaction, the prefixes that are
 // not stored yet, each first seen at its time in prefixes: the IPv4 prefixes
-// first and then the IPv6 prefixes.
+// first and then the IPv6 prefixes. A prefix that does not parse is not stored.
 func (d *Database) UpdatePrefixesBatch(prefixes map[string]time.Time) error {
 	if len(prefixes) == 0 {
 		return nil
@@ -874,6 +874,11 @@ func (d *Database) UpdatePrefixesBatch(prefixes map[string]time.Time) error {
 	prefixesV6 := make(map[string]time.Time)
 
 	for prefix, timestamp := range prefixes {
+		_, err := netip.ParsePrefix(prefix)
+		if err != nil {
+			continue
+		}
+
 		if detectIPVersion(prefix) == ipVersionV4 {
 			prefixesV4[prefix] = timestamp
 		} else {

internal/routewatch/asnfetcher.go

@@ -53,7 +53,8 @@ type ASNFetcher struct {
 	currentInterval  time.Duration
 	consecutiveFails int
 
-	// hourly stats tracking
+	// hourly stats tracking; now is the clock, time.Now except in tests
+	now          func() time.Time
 	statsMu      sync.Mutex
 	successTimes []time.Time
 	errorTimes   []time.Time
@@ -69,6 +70,7 @@ func NewASNFetcher(db database.Store, logger *slog.Logger) *ASNFetcher {
 		stopCh:          make(chan struct{}),
 		done:            make(chan struct{}),
 		currentInterval: baseInterval,
+		now:             time.Now,
 		successTimes:    make([]time.Time, 0),
 		errorTimes:      make([]time.Time, 0),
 	}
@@ -121,7 +123,7 @@ func (f *ASNFetcher) GetStats() server.ASNFetcherStats {
 	f.intervalMu.Unlock()
 
 	// Prune old entries and count
-	cutoff := time.Now().Add(-statsWindow)
+	cutoff := f.now().Add(-statsWindow)
 	f.successTimes = pruneOldTimes(f.successTimes, cutoff)
 	f.errorTimes = pruneOldTimes(f.errorTimes, cutoff)
 
@@ -133,16 +135,15 @@ func (f *ASNFetcher) GetStats() server.ASNFetcherStats {
 	}
 }
 
-// pruneOldTimes removes times older than cutoff and returns the pruned slice.
+// pruneOldTimes drops the times at or before cutoff from the front of times,
+// which holds times in the order they were recorded, and returns the rest. It
+// copies nothing, so it is cheap enough to run on every append.
 func pruneOldTimes(times []time.Time, cutoff time.Time) []time.Time {
-	result := make([]time.Time, 0, len(times))
+	for len(times) > 0 && !times[0].After(cutoff) {
-	for _, t := range times {
+		times = times[1:]
-		if t.After(cutoff) {
-			result = append(result, t)
-		}
 	}
 
-	return result
+	return times
 }
 
 // getInterval returns the current fetch interval.
@@ -170,9 +171,11 @@ func (f *ASNFetcher) recordSuccess() {
 	}
 	f.intervalMu.Unlock()
 
-	// Record success time for stats
+	// Record success time for stats, dropping those older than statsWindow
 	f.statsMu.Lock()
-	f.successTimes = append(f.successTimes, time.Now())
+	now := f.now()
+	f.successTimes = pruneOldTimes(f.successTimes, now.Add(-statsWindow))
+	f.successTimes = append(f.successTimes, now)
 	f.statsMu.Unlock()
 }
 
@@ -194,9 +197,11 @@ func (f *ASNFetcher) recordFailure() {
 	}
 	f.intervalMu.Unlock()
 
-	// Record error time for stats
+	// Record error time for stats, dropping those older than statsWindow
 	f.statsMu.Lock()
-	f.errorTimes = append(f.errorTimes, time.Now())
+	now := f.now()
+	f.errorTimes = pruneOldTimes(f.errorTimes, now.Add(-statsWindow))
+	f.errorTimes = append(f.errorTimes, now)
 	f.statsMu.Unlock()
 }
 

internal/routewatch/asnfetcher_internal_test.go

@@ -203,3 +203,35 @@ func TestASNFetcherStopReturnsByItsDeadline(t *testing.T) {
 		t.Fatal("the fetcher goroutine did not return after the lookup ended")
 	}
 }
+
+// TestRecordKeepsOnlyTheStatsWindow records a success and a failure every
+// second, by a clock the test sets, far longer than statsWindow, and never
+// asks for the stats. The fetcher must keep only the times within the last
+// statsWindow.
+func TestRecordKeepsOnlyTheStatsWindow(t *testing.T) {
+	t.Parallel()
+
+	const (
+		records = 100_000
+		step    = time.Second
+		want    = int(statsWindow / step)
+	)
+
+	clock := time.Now()
+
+	f := NewASNFetcher(newMockStore(), logger.New().Logger)
+	f.now = func() time.Time { return clock }
+
+	for range records {
+		clock = clock.Add(step)
+
+		f.recordSuccess()
+		f.recordFailure()
+	}
+
+	if len(f.successTimes) != want || len(f.errorTimes) != want {
+		t.Errorf("%d success and %d error times kept, want %d of each, "+
+			"those of the last %s", len(f.successTimes), len(f.errorTimes),
+			want, statsWindow)
+	}
+}

internal/routewatch/prefixhandler.go

@@ -169,19 +169,10 @@ func (h *PrefixHandler) HandleMessage(msg *ristypes.RISMessage) {
 	// each peer's last update for a prefix: a message that lists a prefix as
 	// both withdrawn and announced keeps the route (RFC 4271 section 4.3).
 	// A withdrawal names only the prefix and peer: the message's AS path, and
-	// so its origin AS, belongs to the announced prefixes. An update whose
+	// so its origin AS, belongs to the announced prefixes.
-	// prefix does not parse is logged and dropped.
+	for _, prefix := range msg.Withdrawals {
-	for _, withdrawn := range msg.Withdrawals {
-		prefix, err := canonicalPrefix(withdrawn)
-		if err != nil {
-			h.logger.Error("Dropped a withdrawal whose prefix does not parse",
-				"prefix", withdrawn, "peer", msg.Peer, "error", err)
-
-			continue
-		}
-
 		h.batch = append(h.batch, prefixUpdate{
-			prefix:      prefix,
+			prefix:      canonicalPrefix(prefix),
 			peer:        msg.Peer,
 			messageType: "withdrawal",
 			timestamp:   timestamp,
@@ -193,18 +184,9 @@ func (h *PrefixHandler) HandleMessage(msg *ristypes.RISMessage) {
 	}
 
 	for _, announcement := range msg.Announcements {
-		for _, announced := range announcement.Prefixes {
+		for _, prefix := range announcement.Prefixes {
-			prefix, err := canonicalPrefix(announced)
-			if err != nil {
-				h.logger.Error(
-					"Dropped an announcement whose prefix does not parse",
-					"prefix", announced, "peer", msg.Peer, "error", err)
-
-				continue
-			}
-
 			h.batch = append(h.batch, prefixUpdate{
-				prefix:      prefix,
+				prefix:      canonicalPrefix(prefix),
 				originASN:   originASN,
 				peer:        msg.Peer,
 				messageType: "announcement",
@@ -578,15 +560,15 @@ func (h *PrefixHandler) writeBatch(
 // the IP lookup builds when it looks a prefix up. The feed sends IPv6
 // announcements compressed ("2001:db8::/32") but IPv6 withdrawals uncompressed
 // ("2001:db8:0:0:0:0:0:0/32"); stored as received, a withdrawal would not match
-// the route its announcement stored. It returns an error for a prefix that does
+// the route its announcement stored. A prefix that does not parse is returned
-// not parse.
+// unchanged, and the batch flush reports it.
-func canonicalPrefix(prefix string) (string, error) {
+func canonicalPrefix(prefix string) string {
 	p, err := netip.ParsePrefix(prefix)
 	if err != nil {
-		return "", err
+		return prefix
 	}
 
-	return p.Masked().String(), nil
+	return p.Masked().String()
 }
 
 // parseCIDR returns the mask length and the IP version of a prefix string

internal/routewatch/prefixhandler_internal_test.go

@@ -440,12 +440,12 @@ func TestPrefixHandlerRemovesTheRouteOnAnAnnouncementWithoutAnOrigin(
 	}
 }
 
-// TestPrefixHandlerDropsAPrefixThatDoesNotParse runs an announcement of
+// TestPrefixHandlerDoesNotStoreAPrefixThatDoesNotParse runs an announcement of
 // "not-a-prefix" and a withdrawal of "also/bad" through the prefix handler into
 // a real database. Neither is stored in the prefix tables: the prefix count in
 // the stats stays 0, both as kept in memory and as counted from the tables when
 // the database is opened again.
-func TestPrefixHandlerDropsAPrefixThatDoesNotParse(t *testing.T) {
+func TestPrefixHandlerDoesNotStoreAPrefixThatDoesNotParse(t *testing.T) {
 	t.Parallel()
 
 	cfg := &config.Config{StateDir: t.TempDir()}