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

The prefix handler now drops an announcement or withdrawal whose prefix
does not parse, and logs it as an error, so the prefix never reaches the
prefix tables or the prefix count in the stats. Before, no route was
stored for such a prefix, but it was stored as a permanent row in the
prefix tables and counted.

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. A prefix that does not parse is not stored.
+// first and then the IPv6 prefixes.
 func (d *Database) UpdatePrefixesBatch(prefixes map[string]time.Time) error {
 	if len(prefixes) == 0 {
 		return nil
@@ -874,11 +874,6 @@ 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,8 +53,7 @@ type ASNFetcher struct {
 	currentInterval  time.Duration
 	consecutiveFails int
 
-	// hourly stats tracking; now is the clock, time.Now except in tests
+	// hourly stats tracking
-	now          func() time.Time
 	statsMu      sync.Mutex
 	successTimes []time.Time
 	errorTimes   []time.Time
@@ -70,7 +69,6 @@ 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),
 	}
@@ -123,7 +121,7 @@ func (f *ASNFetcher) GetStats() server.ASNFetcherStats {
 	f.intervalMu.Unlock()
 
 	// Prune old entries and count
-	cutoff := f.now().Add(-statsWindow)
+	cutoff := time.Now().Add(-statsWindow)
 	f.successTimes = pruneOldTimes(f.successTimes, cutoff)
 	f.errorTimes = pruneOldTimes(f.errorTimes, cutoff)
 
@@ -135,15 +133,16 @@ func (f *ASNFetcher) GetStats() server.ASNFetcherStats {
 	}
 }
 
-// pruneOldTimes drops the times at or before cutoff from the front of times,
+// pruneOldTimes removes times older than cutoff and returns the pruned slice.
-// 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 {
-	for len(times) > 0 && !times[0].After(cutoff) {
+	result := make([]time.Time, 0, len(times))
-		times = times[1:]
+	for _, t := range times {
+		if t.After(cutoff) {
+			result = append(result, t)
+		}
 	}
 
-	return times
+	return result
 }
 
 // getInterval returns the current fetch interval.
@@ -171,11 +170,9 @@ func (f *ASNFetcher) recordSuccess() {
 	}
 	f.intervalMu.Unlock()
 
-	// Record success time for stats, dropping those older than statsWindow
+	// Record success time for stats
 	f.statsMu.Lock()
-	now := f.now()
+	f.successTimes = append(f.successTimes, time.Now())
-	f.successTimes = pruneOldTimes(f.successTimes, now.Add(-statsWindow))
-	f.successTimes = append(f.successTimes, now)
 	f.statsMu.Unlock()
 }
 
@@ -197,11 +194,9 @@ func (f *ASNFetcher) recordFailure() {
 	}
 	f.intervalMu.Unlock()
 
-	// Record error time for stats, dropping those older than statsWindow
+	// Record error time for stats
 	f.statsMu.Lock()
-	now := f.now()
+	f.errorTimes = append(f.errorTimes, time.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,35 +203,3 @@ 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,10 +169,19 @@ 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.
+	// so its origin AS, belongs to the announced prefixes. An update whose
-	for _, prefix := range msg.Withdrawals {
+	// prefix does not parse is logged and dropped.
+	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:      canonicalPrefix(prefix),
+			prefix:      prefix,
 			peer:        msg.Peer,
 			messageType: "withdrawal",
 			timestamp:   timestamp,
@@ -184,9 +193,18 @@ func (h *PrefixHandler) HandleMessage(msg *ristypes.RISMessage) {
 	}
 
 	for _, announcement := range msg.Announcements {
-		for _, prefix := range announcement.Prefixes {
+		for _, announced := 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:      canonicalPrefix(prefix),
+				prefix:      prefix,
 				originASN:   originASN,
 				peer:        msg.Peer,
 				messageType: "announcement",
@@ -560,15 +578,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. A prefix that does not parse is returned
+// the route its announcement stored. It returns an error for a prefix that does
-// unchanged, and the batch flush reports it.
+// not parse.
-func canonicalPrefix(prefix string) string {
+func canonicalPrefix(prefix string) (string, error) {
 	p, err := netip.ParsePrefix(prefix)
 	if err != nil {
-		return prefix
+		return "", err
 	}
 
-	return p.Masked().String()
+	return p.Masked().String(), nil
 }
 
 // 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(
 	}
 }
 
-// TestPrefixHandlerDoesNotStoreAPrefixThatDoesNotParse runs an announcement of
+// TestPrefixHandlerDropsAPrefixThatDoesNotParse 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 TestPrefixHandlerDoesNotStoreAPrefixThatDoesNotParse(t *testing.T) {
+func TestPrefixHandlerDropsAPrefixThatDoesNotParse(t *testing.T) {
 	t.Parallel()
 
 	cfg := &config.Config{StateDir: t.TempDir()}