From 25c7b87991371cc2b6101c1f40653e245f328289 Mon Sep 17 00:00:00 2001 From: sneak Date: Mon, 26 Oct 2026 23:22:11 +0000 Subject: [PATCH] Never move last seen back or first seen forward (closes #162) Some RIS collectors send updates days old while they catch up on a backlog. Every update to a stored AS, peering or peer set last_seen to the update's time, even an older one, and never lowered first_seen. An AS page could then list a peer first seen after it was last seen. The update statements for asns, peerings and bgp_peers now keep the earlier of the stored and new first_seen and the later of the stored and new last_seen, with MIN and MAX in the SQL. The batch and the single-row writes share each statement. GetOrCreateASN returns the times the row now holds. Route times are unchanged. Model: opus-5-5 --- TODO.md | 4 + internal/database/database.go | 60 +++++--- internal/database/database_internal_test.go | 153 ++++++++++++++++++++ internal/database/models.go | 4 +- 4 files changed, 196 insertions(+), 25 deletions(-) diff --git a/TODO.md b/TODO.md index df0f160..4e987b1 100644 --- a/TODO.md +++ b/TODO.md @@ -33,6 +33,10 @@ tested. # Completed Steps +- 2026-10-26: a write to a stored AS, peering or peer keeps the earlier of the + two first-seen times and the later of the two last-seen times. An old update + from a collector catching up on a backlog no longer moves last seen back + (closes #162) - 2026-10-26: the status page shows "Route Churn/min", the number of prefix announcements and withdrawals, from all peers, in the last full clock minute. `/api/v1/stats` reports it as `route_churn_per_min` in `stream`, which diff --git a/internal/database/database.go b/internal/database/database.go index b62d2a3..8a68944 100644 --- a/internal/database/database.go +++ b/internal/database/database.go @@ -945,6 +945,26 @@ func (d *Database) UpdatePrefixesBatch(prefixes map[string]time.Time) error { return nil } +// These statements update a stored AS, peering or peer. Each keeps the earlier +// of the stored and the new first_seen, and the later of the stored and the new +// last_seen. Some collectors send updates days old while they catch up on a +// backlog. MIN and MAX compare the times as text, which is time order here: the +// driver writes every time in one format, and every time is UTC. +const ( + updateASNSQL = `UPDATE asns + SET first_seen = MIN(first_seen, ?), last_seen = MAX(last_seen, ?) + WHERE asn = ?` + + updatePeeringSQL = `UPDATE peerings + SET first_seen = MIN(first_seen, ?), last_seen = MAX(last_seen, ?) + WHERE as_a = ? AND as_b = ?` + + updatePeerSQL = `UPDATE bgp_peers + SET peer_asn = ?, last_message_type = ?, + first_seen = MIN(first_seen, ?), last_seen = MAX(last_seen, ?) + WHERE peer_ip = ?` +) + // 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. @@ -964,9 +984,8 @@ func newASN(number int, seen SeenTimes) ASN { return asn } -// 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. +// upsertASN stores the seen times of ASN number 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, seen SeenTimes, @@ -979,8 +998,8 @@ func upsertASN( err := selectStmt.QueryRowContext(ctx, number).Scan(&asn.ASN, &handle, &description, &asn.FirstSeen, &asn.LastSeen) if err == nil { - // ASN exists, update last_seen - _, err = updateStmt.ExecContext(ctx, seen.Last, number) + // ASN exists, update its seen times + _, err = updateStmt.ExecContext(ctx, seen.First, seen.Last, number) if err != nil { return false, fmt.Errorf("failed to update ASN %d: %w", number, err) } @@ -1030,8 +1049,7 @@ func (d *Database) GetOrCreateASNBatch(asns map[int]SeenTimes) error { } defer func() { _ = selectStmt.Close() }() - updateStmt, err := tx.PrepareContext(ctx, - "UPDATE asns SET last_seen = ? WHERE asn = ?") + updateStmt, err := tx.PrepareContext(ctx, updateASNSQL) if err != nil { return fmt.Errorf("failed to prepare update statement: %w", err) } @@ -1090,17 +1108,19 @@ func (d *Database) GetOrCreateASN(number int, timestamp time.Time) (*ASN, error) "FROM asns WHERE asn = ?", number). Scan(&asn.ASN, &handle, &description, &asn.FirstSeen, &asn.LastSeen) if err == nil { - // ASN exists, update last_seen + // ASN exists, update its seen times asn.Handle = handle.String asn.Description = description.String - _, err = tx.Exec("UPDATE asns SET last_seen = ? WHERE asn = ?", - timestamp, number) + _, err = tx.Exec(updateASNSQL, timestamp, timestamp, number) if err != nil { return nil, err } - asn.LastSeen = timestamp + // Return the seen times the update stored. + seen := SeenTimes{First: asn.FirstSeen, Last: asn.LastSeen} + seen.Add(timestamp) + asn.FirstSeen, asn.LastSeen = seen.First, seen.Last err = tx.Commit() if err != nil { @@ -1259,9 +1279,7 @@ func (d *Database) RecordPeering(asA, asB int, seen SeenTimes) error { } if exists { - _, err = tx.Exec( - "UPDATE peerings SET last_seen = ? WHERE as_a = ? AND as_b = ?", - seen.Last, asA, asB) + _, err = tx.Exec(updatePeeringSQL, seen.First, seen.Last, asA, asB) } else { _, err = tx.Exec(` INSERT INTO peerings (id, as_a, as_b, first_seen, last_seen) @@ -1316,9 +1334,7 @@ func (d *Database) UpdatePeerBatch(peers map[string]PeerUpdate) error { } defer func() { _ = checkStmt.Close() }() - updateStmt, err := tx.PrepareContext(ctx, - "UPDATE bgp_peers SET peer_asn = ?, last_seen = ?, "+ - "last_message_type = ? WHERE peer_ip = ?") + updateStmt, err := tx.PrepareContext(ctx, updatePeerSQL) if err != nil { return fmt.Errorf("failed to prepare update statement: %w", err) } @@ -1344,7 +1360,8 @@ func (d *Database) UpdatePeerBatch(peers map[string]PeerUpdate) error { if exists { _, err = updateStmt.ExecContext(ctx, update.PeerASN, - update.Seen.Last, update.MessageType, update.PeerIP) + update.MessageType, update.Seen.First, update.Seen.Last, + update.PeerIP) } else { _, err = insertStmt.ExecContext(ctx, generateUUID().String(), update.PeerIP, update.PeerASN, @@ -1393,11 +1410,8 @@ func (d *Database) UpdatePeer( } if exists { - _, err = tx.Exec( - "UPDATE bgp_peers SET peer_asn = ?, last_seen = ?, "+ - "last_message_type = ? WHERE peer_ip = ?", - peerASN, timestamp, messageType, peerIP, - ) + _, err = tx.Exec(updatePeerSQL, + peerASN, messageType, timestamp, timestamp, peerIP) } else { _, err = tx.Exec( "INSERT INTO bgp_peers (id, peer_ip, peer_asn, first_seen, "+ diff --git a/internal/database/database_internal_test.go b/internal/database/database_internal_test.go index e08e82d..f5182d0 100644 --- a/internal/database/database_internal_test.go +++ b/internal/database/database_internal_test.go @@ -20,6 +20,7 @@ import ( "git.eeqj.de/sneak/routewatch/internal/config" "git.eeqj.de/sneak/routewatch/internal/logger" + "git.eeqj.de/sneak/routewatch/internal/ristypes" "git.eeqj.de/sneak/routewatch/pkg/asinfo" "github.com/mattn/go-sqlite3" ) @@ -1960,6 +1961,158 @@ func TestUpdatePeerBatchStoresTheEarliestAndLatestTimesOfANewPeer( } } +// backlogAge is how much older one write is than the other in the tests of +// first_seen and last_seen, as when a collector catches up on a backlog. +const backlogAge = 24 * time.Hour + +// batchSpan is how much earlier a batch's First is than its Last in the tests +// of first_seen and last_seen. +const batchSpan = time.Hour + +// TestWritesKeepEarliestFirstSeenAndLatestLastSeen writes a row of each of +// asns, peerings and bgp_peers through each function that writes one, at a +// newer time and at an older one. Each row takes the older write's First as +// first_seen and the newer write's Last as last_seen. The test writes in both +// orders, each in a database of its own. Newer then older is a collector +// catching up on a backlog; older then newer is a feed that arrives in order. +// Each function writes a row of its own. A batch write's First is batchSpan +// before its Last, so the test catches the two swapped. A single-row write +// takes one time. +func TestWritesKeepEarliestFirstSeenAndLatestLastSeen(t *testing.T) { + t.Parallel() + + const ( + batchASN, singleASN = 64500, 64501 + peerASN = 64502 + asnQuery = "SELECT first_seen, last_seen FROM asns WHERE asn = ?" + peeringQuery = "SELECT first_seen, last_seen FROM peerings " + + "WHERE as_a = ? AND as_b = ?" + peerQuery = "SELECT first_seen, last_seen FROM bgp_peers " + + "WHERE peer_ip = ?" + ) + + for name, olderFirst := range map[string]bool{ + "newer then older": false, + "older then newer": true, + } { + t.Run(name, func(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() }() + + checkSeenTimes(t, db, olderFirst, func(seen SeenTimes) error { + return db.GetOrCreateASNBatch(map[int]SeenTimes{batchASN: seen}) + }, batchSpan, asnQuery, batchASN) + + checkSeenTimes(t, db, olderFirst, func(seen SeenTimes) error { + _, err := db.GetOrCreateASN(singleASN, seen.Last) + + return err + }, 0, asnQuery, singleASN) + + checkSeenTimes(t, db, olderFirst, func(seen SeenTimes) error { + return db.RecordPeering(batchASN, singleASN, seen) + }, batchSpan, peeringQuery, batchASN, singleASN) + + checkSeenTimes(t, db, olderFirst, func(seen SeenTimes) error { + return db.UpdatePeerBatch(map[string]PeerUpdate{ + testV4PeerIP: { + PeerIP: testV4PeerIP, PeerASN: peerASN, + MessageType: ristypes.MessageTypeUpdate, + Seen: seen, + }, + }) + }, batchSpan, peerQuery, testV4PeerIP) + + checkSeenTimes(t, db, olderFirst, func(seen SeenTimes) error { + return db.UpdatePeer(testPeerIP, peerASN, + ristypes.MessageTypeUpdate, seen.Last) + }, 0, peerQuery, testPeerIP) + }) + } +} + +// checkSeenTimes calls write twice, each time with First span before Last: at +// newer times and at times backlogAge older. It writes the older times first +// when olderFirst is true. It checks that query, given args, reads the older +// write's First as first_seen and the newer write's Last as last_seen. +func checkSeenTimes( + t *testing.T, db *Database, olderFirst bool, + write func(seen SeenTimes) error, + span time.Duration, query string, args ...any, +) { + t.Helper() + + newerLast := time.Date(2026, 1, 2, 3, 4, 5, 0, time.UTC) + newer := SeenTimes{First: newerLast.Add(-span), Last: newerLast} + older := SeenTimes{ + First: newer.First.Add(-backlogAge), + Last: newer.Last.Add(-backlogAge), + } + + writes := []SeenTimes{newer, older} + if olderFirst { + writes = []SeenTimes{older, newer} + } + + for _, seen := range writes { + err := write(seen) + if err != nil { + t.Fatalf("writing first %s, last %s: %v", seen.First, seen.Last, err) + } + } + + var firstSeen, lastSeen time.Time + + err := db.db.QueryRowContext(t.Context(), query, args...). + Scan(&firstSeen, &lastSeen) + if err != nil { + t.Fatalf("reading the row: %v", err) + } + + if !firstSeen.Equal(older.First) || !lastSeen.Equal(newer.Last) { + t.Errorf("first seen %s, last seen %s; want %s, %s", + firstSeen, lastSeen, older.First, newer.Last) + } +} + +// TestGetOrCreateASNReturnsStoredSeenTimes stores an AS at a newer time, then +// at an older one. The second call returns the times the row now holds: the +// older first_seen and the newer last_seen. +func TestGetOrCreateASNReturnsStoredSeenTimes(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() }() + + const asn = 64500 + + newer := time.Date(2026, 1, 2, 3, 4, 5, 0, time.UTC) + older := newer.Add(-backlogAge) + + _, err = db.GetOrCreateASN(asn, newer) + if err != nil { + t.Fatalf("GetOrCreateASN at %s: %v", newer, err) + } + + got, err := db.GetOrCreateASN(asn, older) + if err != nil { + t.Fatalf("GetOrCreateASN at %s: %v", older, err) + } + + if !got.FirstSeen.Equal(older) || !got.LastSeen.Equal(newer) { + t.Errorf("first seen %s, last seen %s; want %s, %s", + got.FirstSeen, got.LastSeen, older, newer) + } +} + // 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/models.go b/internal/database/models.go index 9b94d8e..dbb5309 100644 --- a/internal/database/models.go +++ b/internal/database/models.go @@ -125,8 +125,8 @@ type LiveRouteDeletion struct { // 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. +// last seen at Last. A row already stored takes First as its first_seen only +// when First is earlier, and Last as its last_seen only when Last is later. type SeenTimes struct { First time.Time Last time.Time -- 2.54.0