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 _, t := range times {
-		if t.After(cutoff) {
-			result = append(result, t)
-		}
+	for len(times) > 0 && !times[0].After(cutoff) {
+		times = times[1:]
 	}
 
-	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_internal_test.go

@@ -440,6 +440,64 @@ func TestPrefixHandlerRemovesTheRouteOnAnAnnouncementWithoutAnOrigin(
 	}
 }
 
+// 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 TestPrefixHandlerDoesNotStoreAPrefixThatDoesNotParse(t *testing.T) {
+	t.Parallel()
+
+	cfg := &config.Config{StateDir: t.TempDir()}
+
+	db, err := database.New(cfg, logger.New())
+	if err != nil {
+		t.Fatalf("failed to create database: %v", err)
+	}
+
+	// Built without NewPrefixHandler's flush timer, so the test flushes itself.
+	h := &PrefixHandler{db: db, logger: logger.New()}
+
+	withdrawal := &ristypes.RISMessage{
+		Type:            ristypes.MessageTypeUpdate,
+		Peer:            testPeerIP,
+		ParsedTimestamp: time.Date(2026, 1, 2, 3, 4, 5, 0, time.UTC),
+		Withdrawals:     []string{"also/bad"},
+	}
+	h.HandleMessage(peerAnnouncement(testPeerIP, "not-a-prefix"))
+	h.HandleMessage(withdrawal)
+	h.flushBatch()
+
+	checkNoPrefixes := func(when string) {
+		t.Helper()
+
+		stats, err := db.GetStatsContext(context.Background())
+		if err != nil {
+			t.Fatalf("GetStatsContext %s: %v", when, err)
+		}
+
+		if stats.Prefixes != 0 {
+			t.Errorf("prefix count %s = %d, want 0", when, stats.Prefixes)
+		}
+	}
+
+	checkNoPrefixes("after the flush")
+
+	err = db.Close()
+	if err != nil {
+		t.Fatalf("Close: %v", err)
+	}
+
+	db, err = database.New(cfg, logger.New())
+	if err != nil {
+		t.Fatalf("failed to open the database again: %v", err)
+	}
+
+	t.Cleanup(func() { _ = db.Close() })
+
+	checkNoPrefixes("after opening the database again")
+}
+
 // TestStateMessagesGoToThePrefixHandlerNotThePeerHandler checks that STATE
 // messages go to the prefix handler, which acts on a down, and not to the peer
 // handler, which would record a peer whose session never comes up as active.