1 Commits
Author SHA1 Message Date
sneak 00ea3441f7 Keep prefixes that do not parse out of the prefix tables (closes #114)
check / check (push) Canceled after 0s
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
2026-10-20 01:30:51 +00:00
5 changed files with 44 additions and 68 deletions
+1 -6
View File
@@ -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 {
+13 -18
View File
@@ -53,8 +53,7 @@ type ASNFetcher struct {
currentInterval time.Duration
consecutiveFails int
// hourly stats tracking; now is the clock, time.Now except in tests
now func() time.Time
// hourly stats tracking
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,
// 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.
// pruneOldTimes removes times older than cutoff and returns the pruned slice.
func pruneOldTimes(times []time.Time, cutoff time.Time) []time.Time {
for len(times) > 0 && !times[0].After(cutoff) {
times = times[1:]
result := make([]time.Time, 0, len(times))
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 = pruneOldTimes(f.successTimes, now.Add(-statsWindow))
f.successTimes = append(f.successTimes, now)
f.successTimes = append(f.successTimes, time.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 = pruneOldTimes(f.errorTimes, now.Add(-statsWindow))
f.errorTimes = append(f.errorTimes, now)
f.errorTimes = append(f.errorTimes, time.Now())
f.statsMu.Unlock()
}
@@ -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)
}
}
+28 -10
View File
@@ -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.
for _, prefix := range msg.Withdrawals {
// so its origin AS, belongs to the announced prefixes. An update whose
// 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
// unchanged, and the batch flush reports it.
func canonicalPrefix(prefix string) string {
// the route its announcement stored. It returns an error for a prefix that does
// not parse.
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
@@ -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()}