diff --git a/internal/database/counts_internal_test.go b/internal/database/counts_internal_test.go index 76f52be..3fcd7b5 100644 --- a/internal/database/counts_internal_test.go +++ b/internal/database/counts_internal_test.go @@ -148,7 +148,9 @@ func TestLiveCountsTrackWritesInRealtime(t *testing.T) { func writeASNsPrefixesAndPeers(t *testing.T, db *Database, ts time.Time) { t.Helper() - err := db.GetOrCreateASNBatch(map[int]time.Time{64500: ts, 64501: ts}) + seen := SeenTimes{First: ts, Last: ts} + + err := db.GetOrCreateASNBatch(map[int]SeenTimes{64500: seen, 64501: seen}) if err != nil { t.Fatalf("GetOrCreateASNBatch: %v", err) } @@ -164,14 +166,14 @@ func writeASNsPrefixesAndPeers(t *testing.T, db *Database, ts time.Time) { err = db.UpdatePeerBatch(map[string]PeerUpdate{ testV4PeerIP: { PeerIP: testV4PeerIP, PeerASN: 64500, - MessageType: "UPDATE", Timestamp: ts, + MessageType: "UPDATE", Seen: seen, }, }) if err != nil { t.Fatalf("UpdatePeerBatch: %v", err) } - err = db.RecordPeering(64500, 64501, ts) + err = db.RecordPeering(64500, 64501, seen) if err != nil { t.Fatalf("RecordPeering: %v", err) } @@ -192,7 +194,10 @@ func TestLiveCountsSeededFromDatabaseAtStartup(t *testing.T) { ts := time.Date(2026, 1, 2, 3, 4, 5, 0, time.UTC) - err = db.GetOrCreateASNBatch(map[int]time.Time{64500: ts, 64501: ts, 64502: ts}) + seen := SeenTimes{First: ts, Last: ts} + + err = db.GetOrCreateASNBatch( + map[int]SeenTimes{64500: seen, 64501: seen, 64502: seen}) if err != nil { t.Fatalf("GetOrCreateASNBatch: %v", err) } diff --git a/internal/database/database.go b/internal/database/database.go index 16425e9..270d7d1 100644 --- a/internal/database/database.go +++ b/internal/database/database.go @@ -903,13 +903,14 @@ func (d *Database) UpdatePrefixesBatch(prefixes map[string]time.Time) error { return nil } -// newASN returns a new ASN record first and last seen at timestamp, with the -// handle and description from the AS info lookup when it knows the number. -func newASN(number int, timestamp time.Time) ASN { +// newASN returns a new ASN record first seen at seen.First and last seen at +// seen.Last, with the handle and description from the AS info lookup when it +// knows the number. +func newASN(number int, seen SeenTimes) ASN { asn := ASN{ ASN: number, - FirstSeen: timestamp, - LastSeen: timestamp, + FirstSeen: seen.First, + LastSeen: seen.Last, } // Look up ASN info @@ -921,12 +922,12 @@ func newASN(number int, timestamp time.Time) ASN { return asn } -// upsertASN updates the last_seen time of ASN number with the given +// upsertASN sets the last_seen time of ASN number to seen.Last with the given // statements, inserting the ASN when it is new. It reports whether it inserted // it. func upsertASN( ctx context.Context, selectStmt, updateStmt, insertStmt *sql.Stmt, - number int, timestamp time.Time, + number int, seen SeenTimes, ) (bool, error) { var ( asn ASN @@ -937,7 +938,7 @@ func upsertASN( &description, &asn.FirstSeen, &asn.LastSeen) if err == nil { // ASN exists, update last_seen - _, err = updateStmt.ExecContext(ctx, timestamp, number) + _, err = updateStmt.ExecContext(ctx, seen.Last, number) if err != nil { return false, fmt.Errorf("failed to update ASN %d: %w", number, err) } @@ -950,7 +951,7 @@ func upsertASN( } // ASN doesn't exist, create it - asn = newASN(number, timestamp) + asn = newASN(number, seen) _, err = insertStmt.ExecContext(ctx, asn.ASN, asn.Handle, asn.Description, asn.FirstSeen, asn.LastSeen) @@ -962,7 +963,7 @@ func upsertASN( } // GetOrCreateASNBatch creates or updates multiple ASNs in a single transaction -func (d *Database) GetOrCreateASNBatch(asns map[int]time.Time) error { +func (d *Database) GetOrCreateASNBatch(asns map[int]SeenTimes) error { if len(asns) == 0 { return nil } @@ -1004,9 +1005,9 @@ func (d *Database) GetOrCreateASNBatch(asns map[int]time.Time) error { var newASNs int - for number, timestamp := range asns { + for number, seen := range asns { inserted, err := upsertASN(ctx, selectStmt, updateStmt, insertStmt, - number, timestamp) + number, seen) if err != nil { return err } @@ -1075,7 +1076,7 @@ func (d *Database) GetOrCreateASN(number int, timestamp time.Time) (*ASN, error) } // ASN doesn't exist, create it with ASN info lookup - asn = newASN(number, timestamp) + asn = newASN(number, SeenTimes{First: timestamp, Last: timestamp}) _, err = tx.Exec("INSERT INTO asns "+ "(asn, handle, description, first_seen, last_seen) "+ @@ -1182,7 +1183,7 @@ func (d *Database) RecordAnnouncement(announcement *Announcement) error { } // RecordPeering records a peering relationship between two ASNs. -func (d *Database) RecordPeering(asA, asB int, timestamp time.Time) error { +func (d *Database) RecordPeering(asA, asB int, seen SeenTimes) error { // Validate ASNs if asA <= 0 || asB <= 0 { return fmt.Errorf("%w: asA=%d, asB=%d", errInvalidASN, asA, asB) @@ -1218,12 +1219,12 @@ func (d *Database) RecordPeering(asA, asB int, timestamp time.Time) error { if exists { _, err = tx.Exec( "UPDATE peerings SET last_seen = ? WHERE as_a = ? AND as_b = ?", - timestamp, asA, asB) + seen.Last, asA, asB) } else { _, err = tx.Exec(` INSERT INTO peerings (id, as_a, as_b, first_seen, last_seen) VALUES (?, ?, ?, ?, ?)`, - generateUUID().String(), asA, asB, timestamp, timestamp) + generateUUID().String(), asA, asB, seen.First, seen.Last) } if err != nil { @@ -1301,11 +1302,11 @@ func (d *Database) UpdatePeerBatch(peers map[string]PeerUpdate) error { if exists { _, err = updateStmt.ExecContext(ctx, update.PeerASN, - update.Timestamp, update.MessageType, update.PeerIP) + update.Seen.Last, update.MessageType, update.PeerIP) } else { _, err = insertStmt.ExecContext(ctx, generateUUID().String(), update.PeerIP, update.PeerASN, - update.Timestamp, update.Timestamp, update.MessageType) + update.Seen.First, update.Seen.Last, update.MessageType) } if err != nil { diff --git a/internal/database/database_internal_test.go b/internal/database/database_internal_test.go index cd82917..a24f25e 100644 --- a/internal/database/database_internal_test.go +++ b/internal/database/database_internal_test.go @@ -615,10 +615,12 @@ func TestBatchWriteDuringCheckpoint(t *testing.T) { }) ts := time.Now().UTC() + seen := SeenTimes{First: ts, Last: ts} + for i := range contentionIterations { - asns := map[int]time.Time{ - i % contendedASNCount: ts, - (i % contendedASNCount) + asnSecondBand: ts, + asns := map[int]SeenTimes{ + i % contendedASNCount: seen, + (i % contendedASNCount) + asnSecondBand: seen, } err := db.GetOrCreateASNBatch(asns) @@ -747,7 +749,8 @@ func TestBatchWriteWaitsForAnalyze(t *testing.T) { db, messages := newDatabaseWithLockMessages(t) ctx := context.Background() - asns := map[int]time.Time{lockTestASN: time.Now().UTC()} + now := time.Now().UTC() + asns := map[int]SeenTimes{lockTestASN: {First: now, Last: now}} // A stored AS gives ANALYZE a table to analyze and statistics to write. err := db.GetOrCreateASNBatch(asns) @@ -1663,6 +1666,48 @@ func TestUpdatePrefixesBatchKeepsFirstSeen(t *testing.T) { } } +// TestUpdatePeerBatchStoresTheEarliestAndLatestTimesOfANewPeer writes a new +// peer whose updates span 5 seconds. Its row is first seen at the earliest time +// and last seen at the latest. +func TestUpdatePeerBatchStoresTheEarliestAndLatestTimesOfANewPeer( + t *testing.T, +) { + t.Parallel() + + db, err := New(&config.Config{StateDir: t.TempDir()}, logger.New()) + if err != nil { + t.Fatalf("failed to create database: %v", err) + } + defer func() { _ = db.Close() }() + + first := time.Date(2026, 1, 2, 3, 4, 5, 0, time.UTC) + last := first.Add(5 * time.Second) + + err = db.UpdatePeerBatch(map[string]PeerUpdate{ + testPeerIP: { + PeerIP: testPeerIP, PeerASN: 64500, MessageType: "UPDATE", + Seen: SeenTimes{First: first, Last: last}, + }, + }) + if err != nil { + t.Fatalf("UpdatePeerBatch: %v", err) + } + + var firstSeen, lastSeen time.Time + + err = db.db.QueryRowContext(t.Context(), + "SELECT first_seen, last_seen FROM bgp_peers WHERE peer_ip = ?", + testPeerIP).Scan(&firstSeen, &lastSeen) + if err != nil { + t.Fatalf("reading peer %s: %v", testPeerIP, err) + } + + if !firstSeen.Equal(first) || !lastSeen.Equal(last) { + t.Errorf("peer %s: first seen %s, last seen %s; want %s, %s", + testPeerIP, firstSeen, lastSeen, first, last) + } +} + // TestUpsertLiveRouteBatchWritesInKeyOrder writes new routes of two peers out // of key order, the last one a second announcement of the first, and checks // that the rows were inserted in key order, peer then prefix (each new row's id diff --git a/internal/database/interface.go b/internal/database/interface.go index 2f5f8ac..1c49b10 100644 --- a/internal/database/interface.go +++ b/internal/database/interface.go @@ -34,7 +34,7 @@ type Stats struct { type Store interface { // ASN operations GetOrCreateASN(number int, timestamp time.Time) (*ASN, error) - GetOrCreateASNBatch(asns map[int]time.Time) error + GetOrCreateASNBatch(asns map[int]SeenTimes) error // Prefix operations GetOrCreatePrefix(prefix string, timestamp time.Time) (*Prefix, error) @@ -44,7 +44,7 @@ type Store interface { RecordAnnouncement(announcement *Announcement) error // Peering operations - RecordPeering(asA, asB int, timestamp time.Time) error + RecordPeering(asA, asB int, seen SeenTimes) error // Statistics GetStats() (Stats, error) diff --git a/internal/database/models.go b/internal/database/models.go index 5392a4c..8797ae7 100644 --- a/internal/database/models.go +++ b/internal/database/models.go @@ -122,12 +122,35 @@ type LiveRouteDeletion struct { IPVersion int } -// PeerUpdate represents parameters for updating a peer +// SeenTimes is the earliest and the latest time one batch of updates saw an AS, +// a peer or a peering. A row the batch adds is stored first seen at First and +// last seen at Last; a row already stored keeps its first_seen and has its +// last_seen set to Last. +type SeenTimes struct { + First time.Time + Last time.Time +} + +// Add makes t the earliest time of s when it is earlier than First, and the +// latest when it is later than Last. The zero SeenTimes holds no time yet, so +// the first Add sets both. +func (s *SeenTimes) Add(t time.Time) { + if s.First.IsZero() || t.Before(s.First) { + s.First = t + } + + if s.Last.IsZero() || t.After(s.Last) { + s.Last = t + } +} + +// PeerUpdate represents parameters for updating a peer. PeerASN and +// MessageType are those of the peer's latest update. type PeerUpdate struct { PeerIP string PeerASN int MessageType string - Timestamp time.Time + Seen SeenTimes } // ASNWHOISUpdate contains WHOIS data for updating an ASN record. diff --git a/internal/routewatch/app_integration_internal_test.go b/internal/routewatch/app_integration_internal_test.go index 8cb2474..f467126 100644 --- a/internal/routewatch/app_integration_internal_test.go +++ b/internal/routewatch/app_integration_internal_test.go @@ -152,7 +152,7 @@ func (m *mockStore) RecordAnnouncement(_ *database.Announcement) error { } // RecordPeering mock implementation -func (m *mockStore) RecordPeering(asA, asB int, _ time.Time) error { +func (m *mockStore) RecordPeering(asA, asB int, _ database.SeenTimes) error { m.mu.Lock() defer m.mu.Unlock() @@ -396,16 +396,16 @@ func (m *mockStore) DeleteLiveRoutesOfPeer(_ string) (int, error) { } // GetOrCreateASNBatch mock implementation -func (m *mockStore) GetOrCreateASNBatch(asns map[int]time.Time) error { +func (m *mockStore) GetOrCreateASNBatch(asns map[int]database.SeenTimes) error { m.mu.Lock() defer m.mu.Unlock() - for number, timestamp := range asns { + for number, seen := range asns { if _, exists := m.ASNs[number]; !exists { m.ASNs[number] = &database.ASN{ ASN: number, - FirstSeen: timestamp, - LastSeen: timestamp, + FirstSeen: seen.First, + LastSeen: seen.Last, } m.ASNCount++ } @@ -766,7 +766,9 @@ const ( callClose = "close" ) -func (s *closeOrderStore) GetOrCreateASNBatch(asns map[int]time.Time) error { +func (s *closeOrderStore) GetOrCreateASNBatch( + asns map[int]database.SeenTimes, +) error { s.record(callWrite) return s.Store.GetOrCreateASNBatch(asns) diff --git a/internal/routewatch/ashandler.go b/internal/routewatch/ashandler.go index 3991c80..d82b9f8 100644 --- a/internal/routewatch/ashandler.go +++ b/internal/routewatch/ashandler.go @@ -229,19 +229,21 @@ func (h *ASHandler) flushBatch() { return } - // Process ASNs first (deduped) - asnMap := make(map[int]time.Time) + // The earliest and the latest time the batch saw each AS. The batch holds + // several updates for an AS, from many peers, so a new AS is stored first + // seen at the earliest of them and last seen at the latest. + asns := make(map[int]database.SeenTimes) for _, op := range batch { - if existing, ok := asnMap[op.number]; !ok || op.timestamp.After(existing) { - asnMap[op.number] = op.timestamp - } + seen := asns[op.number] + seen.Add(op.timestamp) + asns[op.number] = seen } // Process all ASNs in a single batch transaction - err := h.db.GetOrCreateASNBatch(asnMap) + err := h.db.GetOrCreateASNBatch(asns) if err != nil { h.logger.Error("Failed to process ASN batch", - "error", err, "count", len(asnMap)) + "error", err, "count", len(asns)) } h.mu.Lock() diff --git a/internal/routewatch/ashandler_internal_test.go b/internal/routewatch/ashandler_internal_test.go new file mode 100644 index 0000000..8510a7a --- /dev/null +++ b/internal/routewatch/ashandler_internal_test.go @@ -0,0 +1,44 @@ +package routewatch + +import ( + "testing" + + "git.eeqj.de/sneak/routewatch/internal/logger" +) + +// TestASHandlerStoresTheEarliestAndLatestTimesOfANewAS handles two updates +// whose path carries an AS the database does not hold yet, the second 5 +// seconds after the first, in one batch into a real database. The AS is stored +// first seen at the time of the first update and last seen at the time of the +// second. +func TestASHandlerStoresTheEarliestAndLatestTimesOfANewAS(t *testing.T) { + t.Parallel() + + db := newTestDatabase(t) + + // Built without NewASHandler's flush goroutine, so the test flushes itself. + h := &ASHandler{db: db, logger: logger.New()} + + first := peerAnnouncement(testPeerIP, testV4Prefix) + second := peerAnnouncement(testPeerIP, testV4Prefix) + second.ParsedTimestamp = first.ParsedTimestamp.Add(secondUpdateAfter) + + h.HandleMessage(first) + h.HandleMessage(second) + h.flushBatch() + + asn, _, err := db.GetASDetails(testASNA) + if err != nil { + t.Fatalf("GetASDetails(%d): %v", testASNA, err) + } + + if !asn.FirstSeen.Equal(first.ParsedTimestamp) { + t.Errorf("first_seen of AS%d = %s, want %s", testASNA, + asn.FirstSeen, first.ParsedTimestamp) + } + + if !asn.LastSeen.Equal(second.ParsedTimestamp) { + t.Errorf("last_seen of AS%d = %s, want %s", testASNA, + asn.LastSeen, second.ParsedTimestamp) + } +} diff --git a/internal/routewatch/batchwrite_internal_test.go b/internal/routewatch/batchwrite_internal_test.go index 605a8d9..b9ce48d 100644 --- a/internal/routewatch/batchwrite_internal_test.go +++ b/internal/routewatch/batchwrite_internal_test.go @@ -66,7 +66,7 @@ func (s *blockingStore) UpsertLiveRouteBatch(routes []*database.LiveRoute) error return nil } -func (s *blockingStore) GetOrCreateASNBatch(asns map[int]time.Time) error { +func (s *blockingStore) GetOrCreateASNBatch(asns map[int]database.SeenTimes) error { s.waitForRelease() s.mu.Lock() defer s.mu.Unlock() @@ -96,10 +96,12 @@ func (s *blockingStore) DeleteLiveRoutesOfPeer(peerIP string) (int, error) { return s.mockStore.DeleteLiveRoutesOfPeer(peerIP) } -func (s *blockingStore) RecordPeering(asA, asB int, ts time.Time) error { +func (s *blockingStore) RecordPeering( + asA, asB int, seen database.SeenTimes, +) error { s.waitForRelease() - return s.mockStore.RecordPeering(asA, asB, ts) + return s.mockStore.RecordPeering(asA, asB, seen) } func (s *blockingStore) waitForRelease() { diff --git a/internal/routewatch/peerhandler.go b/internal/routewatch/peerhandler.go index 7854695..6fb3278 100644 --- a/internal/routewatch/peerhandler.go +++ b/internal/routewatch/peerhandler.go @@ -37,8 +37,8 @@ const ( // PeerHandler tracks BGP peers from UPDATE messages using batched operations. // It maintains a queue of peer updates and periodically flushes them to the database -// in batches to improve performance. The handler deduplicates peer updates within -// each batch, keeping only the most recent update for each peer IP address. +// in batches to improve performance. The handler writes one row per peer IP +// address for each batch, with the earliest and the latest time of its updates. type PeerHandler struct { db database.Store logger *logger.Logger @@ -227,31 +227,28 @@ func (h *PeerHandler) flushBatch() { return } - // Deduplicate by peer IP, keeping the latest update for each peer - peerMap := make(map[string]peerUpdate) + // The earliest and the latest time the batch saw each peer, with the peer + // AS and message type of its latest update. The batch holds several + // updates for a peer, so a new peer is stored first seen at the earliest of + // them and last seen at the latest. + peers := make(map[string]database.PeerUpdate) for _, update := range batch { - existing, ok := peerMap[update.peerIP] - if !ok || update.timestamp.After(existing.timestamp) { - peerMap[update.peerIP] = update + peer, ok := peers[update.peerIP] + if !ok || update.timestamp.After(peer.Seen.Last) { + peer.PeerIP = update.peerIP + peer.PeerASN = update.peerASN + peer.MessageType = update.messageType } - } - // Convert to database format - dbPeerMap := make(map[string]database.PeerUpdate) - for peerIP, update := range peerMap { - dbPeerMap[peerIP] = database.PeerUpdate{ - PeerIP: update.peerIP, - PeerASN: update.peerASN, - MessageType: update.messageType, - Timestamp: update.timestamp, - } + peer.Seen.Add(update.timestamp) + peers[update.peerIP] = peer } // Process all peers in a single batch transaction - err := h.db.UpdatePeerBatch(dbPeerMap) + err := h.db.UpdatePeerBatch(peers) if err != nil { h.logger.Error("Failed to process peer batch", - "error", err, "count", len(dbPeerMap)) + "error", err, "count", len(peers)) } h.mu.Lock() diff --git a/internal/routewatch/peerhandler_internal_test.go b/internal/routewatch/peerhandler_internal_test.go new file mode 100644 index 0000000..23e351f --- /dev/null +++ b/internal/routewatch/peerhandler_internal_test.go @@ -0,0 +1,59 @@ +package routewatch + +import ( + "testing" + + "git.eeqj.de/sneak/routewatch/internal/database" + "git.eeqj.de/sneak/routewatch/internal/logger" +) + +// peerBatchStore keeps the peers of the last UpdatePeerBatch call. The +// database has no read of the peer table, so the peer handler's test checks +// what the handler writes. +type peerBatchStore struct { + *mockStore + + peers map[string]database.PeerUpdate +} + +func (s *peerBatchStore) UpdatePeerBatch( + peers map[string]database.PeerUpdate, +) error { + s.peers = peers + + return nil +} + +// TestPeerHandlerWritesTheEarliestAndLatestTimesOfANewPeer handles two updates +// from one peer, the second 5 seconds after the first, in one batch. The peer +// handler writes the peer first seen at the time of the first update and last +// seen at the time of the second. +func TestPeerHandlerWritesTheEarliestAndLatestTimesOfANewPeer(t *testing.T) { + t.Parallel() + + store := &peerBatchStore{mockStore: newMockStore()} + + // Built without NewPeerHandler's flush goroutine, so the test flushes + // itself. + h := &PeerHandler{db: store, logger: logger.New()} + + first := peerAnnouncement(testPeerIP, testV4Prefix) + second := peerAnnouncement(testPeerIP, testV4Prefix) + second.ParsedTimestamp = first.ParsedTimestamp.Add(secondUpdateAfter) + + h.HandleMessage(first) + h.HandleMessage(second) + h.flushBatch() + + seen := store.peers[testPeerIP].Seen + + if !seen.First.Equal(first.ParsedTimestamp) { + t.Errorf("first_seen of peer %s = %s, want %s", testPeerIP, + seen.First, first.ParsedTimestamp) + } + + if !seen.Last.Equal(second.ParsedTimestamp) { + t.Errorf("last_seen of peer %s = %s, want %s", testPeerIP, + seen.Last, second.ParsedTimestamp) + } +} diff --git a/internal/routewatch/peeringhandler.go b/internal/routewatch/peeringhandler.go index f304082..b1cddc4 100644 --- a/internal/routewatch/peeringhandler.go +++ b/internal/routewatch/peeringhandler.go @@ -42,10 +42,11 @@ type PeeringHandler struct { db database.Store logger *logger.Logger - // In-memory AS path tracking + // In-memory AS path tracking: each path, JSON-encoded, with the earliest + // and the latest time an update carried it. mu sync.Mutex - asPaths map[string]time.Time // key is JSON-encoded AS path - droppedPaths int // paths dropped because the map was full + asPaths map[string]database.SeenTimes + droppedPaths int // paths dropped because the map was full // writing is how many paths processPeerings took from asPaths and has // not finished writing the peerings of. writing int @@ -64,7 +65,7 @@ func NewPeeringHandler(db database.Store, logger *logger.Logger) *PeeringHandler h := &PeeringHandler{ db: db, logger: logger, - asPaths: make(map[string]time.Time), + asPaths: make(map[string]database.SeenTimes), stopCh: make(chan context.Context, 1), done: make(chan struct{}), } @@ -118,16 +119,18 @@ func (h *PeeringHandler) HandleMessage(msg *ristypes.RISMessage) { key := string(pathJSON) h.mu.Lock() - if _, exists := h.asPaths[key]; exists { - // Already tracked: refresh its timestamp. - h.asPaths[key] = timestamp - } else if len(h.asPaths) >= maxTrackedPaths { + defer h.mu.Unlock() + + seen, exists := h.asPaths[key] + if !exists && len(h.asPaths) >= maxTrackedPaths { // Map is full; drop this new path and count it. h.droppedPaths++ - } else { - h.asPaths[key] = timestamp + + return } - h.mu.Unlock() + + seen.Add(timestamp) + h.asPaths[key] = seen } // ProcessPeeringsNow triggers immediate processing of all accumulated AS @@ -190,7 +193,7 @@ func (h *PeeringHandler) processPeerings() { // released, so the map never grows past a single interval's traffic. h.mu.Lock() paths := h.asPaths - h.asPaths = make(map[string]time.Time) + h.asPaths = make(map[string]database.SeenTimes) h.writing = len(paths) dropped := h.droppedPaths h.mu.Unlock() @@ -204,9 +207,13 @@ func (h *PeeringHandler) processPeerings() { low, high int } - peerings := make(map[peeringKey]time.Time) + // The earliest and the latest time an update carried each peering. Several + // paths, each carried at several times, can hold a peering, so a new + // peering is stored first seen at the earliest of them and last seen at the + // latest. + peerings := make(map[peeringKey]database.SeenTimes) - for pathJSON, timestamp := range paths { + for pathJSON, pathSeen := range paths { var path []int err := json.Unmarshal([]byte(pathJSON), &path) @@ -233,10 +240,11 @@ func (h *PeeringHandler) processPeerings() { } key := peeringKey{low: low, high: high} - // Update timestamp if this is newer - if existing, ok := peerings[key]; !ok || timestamp.After(existing) { - peerings[key] = timestamp - } + + seen := peerings[key] + seen.Add(pathSeen.First) + seen.Add(pathSeen.Last) + peerings[key] = seen } } @@ -244,8 +252,8 @@ func (h *PeeringHandler) processPeerings() { start := time.Now() successCount := 0 - for key, ts := range peerings { - err := h.db.RecordPeering(key.low, key.high, ts) + for key, seen := range peerings { + err := h.db.RecordPeering(key.low, key.high, seen) if err != nil { h.logger.Error("Failed to record peering", "as_a", key.low, diff --git a/internal/routewatch/peeringhandler_internal_test.go b/internal/routewatch/peeringhandler_internal_test.go index b11737a..75bc05a 100644 --- a/internal/routewatch/peeringhandler_internal_test.go +++ b/internal/routewatch/peeringhandler_internal_test.go @@ -31,12 +31,14 @@ type recordingStore struct { calls int } -func (r *recordingStore) RecordPeering(asA, asB int, ts time.Time) error { +func (r *recordingStore) RecordPeering( + asA, asB int, seen database.SeenTimes, +) error { r.mu.Lock() r.calls++ r.mu.Unlock() - return r.mockStore.RecordPeering(asA, asB, ts) + return r.mockStore.RecordPeering(asA, asB, seen) } func (r *recordingStore) callCount() int { @@ -52,7 +54,7 @@ func newTestHandler(db database.Store) *PeeringHandler { return &PeeringHandler{ db: db, logger: logger.New(), - asPaths: make(map[string]time.Time), + asPaths: make(map[string]database.SeenTimes), } } @@ -220,10 +222,10 @@ func TestPeeringHandlerCapDropsAndCounts(t *testing.T) { existing := pathKey(t, testASNA, testASNB) h.mu.Lock() - h.asPaths[existing] = now + h.asPaths[existing] = database.SeenTimes{First: now, Last: now} for i := 0; len(h.asPaths) < maxTrackedPaths; i++ { - h.asPaths[strconv.Itoa(i)] = now + h.asPaths[strconv.Itoa(i)] = database.SeenTimes{First: now, Last: now} } h.mu.Unlock() @@ -253,11 +255,53 @@ func TestPeeringHandlerCapDropsAndCounts(t *testing.T) { } h.mu.Lock() - gotTS := h.asPaths[existing] + gotSeen := h.asPaths[existing] h.mu.Unlock() - if !gotTS.Equal(refreshed) { - t.Fatalf("existing path timestamp not refreshed: want %v, got %v", - refreshed, gotTS) + if !gotSeen.Last.Equal(refreshed) { + t.Fatalf("existing path last seen not refreshed: want %v, got %v", + refreshed, gotSeen.Last) + } +} + +// TestPeeringHandlerStoresTheEarliestAndLatestTimesOfANewPeering handles two +// updates whose path carries a peering the database does not hold yet, the +// second 5 seconds after the first, in one run into a real database. The +// peering is stored first seen at the time of the first update and last seen +// at the time of the second. +func TestPeeringHandlerStoresTheEarliestAndLatestTimesOfANewPeering( + t *testing.T, +) { + t.Parallel() + + db := newTestDatabase(t) + h := newTestHandler(db) + + first := peerAnnouncement(testPeerIP, testV4Prefix) + second := peerAnnouncement(testPeerIP, testV4Prefix) + second.ParsedTimestamp = first.ParsedTimestamp.Add(secondUpdateAfter) + + h.HandleMessage(first) + h.HandleMessage(second) + h.ProcessPeeringsNow() + + peers, err := db.GetASPeers(testASNA) + if err != nil { + t.Fatalf("GetASPeers(%d): %v", testASNA, err) + } + + if len(peers) != 1 || peers[0].ASN != testASNB { + t.Fatalf("peers of AS%d = %+v, want AS%d alone", testASNA, peers, + testASNB) + } + + if !peers[0].FirstSeen.Equal(first.ParsedTimestamp) { + t.Errorf("first_seen of peering AS%d-AS%d = %s, want %s", testASNA, + testASNB, peers[0].FirstSeen, first.ParsedTimestamp) + } + + if !peers[0].LastSeen.Equal(second.ParsedTimestamp) { + t.Errorf("last_seen of peering AS%d-AS%d = %s, want %s", testASNA, + testASNB, peers[0].LastSeen, second.ParsedTimestamp) } } diff --git a/internal/routewatch/prefixhandler.go b/internal/routewatch/prefixhandler.go index 329651e..e7f62f9 100644 --- a/internal/routewatch/prefixhandler.go +++ b/internal/routewatch/prefixhandler.go @@ -347,8 +347,19 @@ func (h *PrefixHandler) flushBatch() { lastUpdates := make(map[prefixAndPeer]prefixUpdate) + // Each prefix the batch names, announced or withdrawn, with the earliest + // timestamp among all of its updates, not only each peer's last one: the + // prefix tables store only first_seen, written when the prefix is new and + // never changed after. + prefixesToUpdate := make(map[string]time.Time) + for _, update := range batch { lastUpdates[prefixAndPeer{update.prefix, update.peer}] = update + + seen, exists := prefixesToUpdate[update.prefix] + if !exists || update.timestamp.Before(seen) { + prefixesToUpdate[update.prefix] = update.timestamp + } } // Collect routes to upsert and delete @@ -357,16 +368,7 @@ func (h *PrefixHandler) flushBatch() { routesToDelete []database.LiveRouteDeletion ) - // Collect unique prefixes to update - prefixesToUpdate := make(map[string]time.Time) - for _, update := range lastUpdates { - // Track prefix for both announcements and withdrawals - seen, exists := prefixesToUpdate[update.prefix] - if !exists || update.timestamp.After(seen) { - prefixesToUpdate[update.prefix] = update.timestamp - } - // An announcement is stored when its path gives an origin, or when it // ends in an AS set of several members: that route is stored with // origin 0 (none). An announcement with an empty path is not stored. diff --git a/internal/routewatch/prefixhandler_internal_test.go b/internal/routewatch/prefixhandler_internal_test.go index 12aeb9d..30c6433 100644 --- a/internal/routewatch/prefixhandler_internal_test.go +++ b/internal/routewatch/prefixhandler_internal_test.go @@ -26,6 +26,10 @@ const ( testV6Prefix = "2001:db8:1::/48" ) +// secondUpdateAfter is how long after the first update the tests of first_seen +// and last_seen send the second, in the same batch. +const secondUpdateAfter = 5 * time.Second + // TestPrefixHandlerStoresPrefixesTheIPLookupFinds runs announcements and // withdrawals through the prefix handler into a real database and looks the // addresses up. The prefixes are written the way the feed sends them: IPv6 @@ -83,6 +87,49 @@ func TestPrefixHandlerStoresPrefixesTheIPLookupFinds(t *testing.T) { } } +// TestPrefixHandlerStoresANewPrefixFirstSeenAtItsEarliestUpdate announces a +// prefix the database does not hold yet twice in one batch, the second time 5 +// seconds after the first, through the prefix handler into a real database. The +// prefix is stored first seen at the time of the first announcement, whether +// the two come from two peers or from one, and the IP lookup answers with that +// time. +func TestPrefixHandlerStoresANewPrefixFirstSeenAtItsEarliestUpdate( + t *testing.T, +) { + t.Parallel() + + const otherPeerIP = "2001:db8:ffff::2" + + tests := []struct { + name string + firstPeer, secondPeer string + }{ + {name: "two peers", firstPeer: testPeerIP, secondPeer: otherPeerIP}, + {name: "one peer", firstPeer: testPeerIP, secondPeer: testPeerIP}, + } + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + t.Parallel() + + first := peerAnnouncement(tt.firstPeer, testV4Prefix) + second := peerAnnouncement(tt.secondPeer, testV4Prefix) + second.ParsedTimestamp = first.ParsedTimestamp.Add(secondUpdateAfter) + db := handleBatches(t, [][]*ristypes.RISMessage{{first, second}}) + + info, err := db.GetIPInfo("192.0.2.1") + if err != nil { + t.Fatalf("GetIPInfo(192.0.2.1): %v", err) + } + + if !info.FirstSeen.Equal(first.ParsedTimestamp) { + t.Errorf("first_seen of %s = %s, want %s", testV4Prefix, + info.FirstSeen, first.ParsedTimestamp) + } + }) + } +} + // TestPrefixHandlerAppliesEachPeersLastUpdateInABatch runs updates for one // prefix through the prefix handler into a real database. All updates carry the // same timestamp, as updates within one second of the feed do. Within a batch,