package database import ( "context" "database/sql" "encoding/json" "errors" "fmt" "net/netip" "time" "github.com/bits-and-blooms/bitset" "git.eeqj.de/sneak/routewatch/internal/routetable" ) // keptRoutesPerRow is how many kept routes a row of the routes table has // columns for: the newest and the older one. const keptRoutesPerRow = 2 // errUnknownRoutePeer is returned for a kept route whose peer has no number in // route_peers. var errUnknownRoutePeer = errors.New("kept route of a peer without a number") // The statements of the routes and route_peers tables (schema.sql). const ( insertRoutePeerSQL = `INSERT INTO route_peers (number, address) VALUES (?, ?)` routePeerNumberSQL = `SELECT number FROM route_peers WHERE address = ?` routePeersSQL = `SELECT number, address FROM route_peers` upsertRouteSQL = `INSERT INTO routes (prefix, origin_asn, peers, newest_peer, newest_as_path, newest_time, older_peer, older_as_path, older_time) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?) ON CONFLICT (prefix, origin_asn) DO UPDATE SET peers = excluded.peers, newest_peer = excluded.newest_peer, newest_as_path = excluded.newest_as_path, newest_time = excluded.newest_time, older_peer = excluded.older_peer, older_as_path = excluded.older_as_path, older_time = excluded.older_time` deleteRouteSQL = `DELETE FROM routes WHERE prefix = ? AND origin_asn = ?` routesSQL = `SELECT prefix, origin_asn, peers, newest_peer, newest_as_path, newest_time, older_peer, older_as_path, older_time FROM routes` ) // WriteRouteChanges writes, in one transaction, what one call of the routing // table's TakeChanges returned. It stores each new peer with its number, writes // the row of each changed prefix and origin AS pair, and deletes the row of // each removed one. Route times are stored to the second, as the feed gives // them. // // When it fails, it writes nothing. The caller then writes these changes and // peers again, because later changes can name the peers it did not store. func (d *Database) WriteRouteChanges( changes []routetable.Change, newPeers map[string]uint, ) error { if len(changes) == 0 && len(newPeers) == 0 { return nil } d.lock("WriteRouteChanges") defer d.unlock() tx, err := d.beginTx() if err != nil { return fmt.Errorf("failed to begin transaction: %w", err) } defer d.rollback(tx) ctx := context.Background() // The new peers first, so that the rows can look up their numbers. for address, number := range newPeers { _, err := tx.ExecContext(ctx, insertRoutePeerSQL, number, address) if err != nil { return fmt.Errorf("failed to store peer %s: %w", address, err) } } err = writeRouteRows(ctx, tx, changes) if err != nil { return err } err = tx.Commit() if err != nil { return fmt.Errorf("failed to commit transaction: %w", err) } return nil } // writeRouteRows upserts the row of each change in tx, or deletes it for a // removed pair. func writeRouteRows( ctx context.Context, tx *loggingTx, changes []routetable.Change, ) error { upsert, err := tx.PrepareContext(ctx, upsertRouteSQL) if err != nil { return fmt.Errorf("failed to prepare route upsert: %w", err) } defer func() { _ = upsert.Close() }() del, err := tx.PrepareContext(ctx, deleteRouteSQL) if err != nil { return fmt.Errorf("failed to prepare route delete: %w", err) } defer func() { _ = del.Close() }() peerNumber, err := tx.PrepareContext(ctx, routePeerNumberSQL) if err != nil { return fmt.Errorf("failed to prepare peer number lookup: %w", err) } defer func() { _ = peerNumber.Close() }() for _, change := range changes { prefix := change.Prefix.String() if change.Removed { _, err := del.ExecContext(ctx, prefix, change.OriginASN) if err != nil { return fmt.Errorf("failed to delete route %s: %w", prefix, err) } continue } row, err := routeRow(ctx, peerNumber, change) if err != nil { return fmt.Errorf("failed to write route %s: %w", prefix, err) } _, err = upsert.ExecContext(ctx, row...) if err != nil { return fmt.Errorf("failed to write route %s: %w", prefix, err) } } return nil } // routeRow returns the values of the routes row of change, in the order of // upsertRouteSQL. peerNumber looks up a peer's number by its address. The // table keeps at most keptRoutesPerRow routes of a pair. func routeRow( ctx context.Context, peerNumber *sql.Stmt, change routetable.Change, ) ([]any, error) { peers, err := change.Peers.MarshalBinary() if err != nil { return nil, err } row := []any{change.Prefix.String(), change.OriginASN, peers} for i := range keptRoutesPerRow { if i >= len(change.Routes) { row = append(row, nil, nil, nil) continue } r := change.Routes[i] var number uint err := peerNumber.QueryRowContext(ctx, r.Peer).Scan(&number) if errors.Is(err, sql.ErrNoRows) { return nil, fmt.Errorf("%w: %s", errUnknownRoutePeer, r.Peer) } if err != nil { return nil, err } path, err := json.Marshal(r.ASPath) if err != nil { return nil, err } row = append(row, number, string(path), r.Time.Unix()) } return row, nil } // ReadRouteTable returns the routing table that the routes and route_peers // tables hold, as startup reads it. It reads the peers first, so that the // table keeps their numbers, then loads each row into the table as it reads // it. The two reads are separate, so nothing may write the tables meanwhile. func (d *Database) ReadRouteTable(ctx context.Context) (*routetable.Table, error) { peerNumbers, err := d.readRoutePeers(ctx) if err != nil { return nil, err } addresses := make(map[uint]string, len(peerNumbers)) for address, number := range peerNumbers { addresses[number] = address } table := routetable.New(routetable.Params{PeerNumbers: peerNumbers}) rows, err := d.db.QueryContext(ctx, routesSQL) if err != nil { return nil, fmt.Errorf("failed to read routes: %w", err) } defer func() { _ = rows.Close() }() for rows.Next() { change, err := scanRouteRow(rows, addresses) if err != nil { return nil, err } table.Load(change) } err = rows.Err() if err != nil { return nil, fmt.Errorf("failed to read routes: %w", err) } return table, nil } // readRoutePeers returns the number of each peer in route_peers, by address. func (d *Database) readRoutePeers(ctx context.Context) (map[string]uint, error) { rows, err := d.db.QueryContext(ctx, routePeersSQL) if err != nil { return nil, fmt.Errorf("failed to read route peers: %w", err) } defer func() { _ = rows.Close() }() peerNumbers := make(map[string]uint) for rows.Next() { var ( number uint address string ) err := rows.Scan(&number, &address) if err != nil { return nil, fmt.Errorf("failed to read a route peer: %w", err) } peerNumbers[address] = number } err = rows.Err() if err != nil { return nil, fmt.Errorf("failed to read route peers: %w", err) } return peerNumbers, nil } // keptRouteColumns are the three columns of one kept route in a routes row. type keptRouteColumns struct { peer sql.Null[uint] asPath sql.NullString time sql.NullInt64 } // scanRouteRow returns the content of the routes row rows is at, with each kept // route's peer named by its address in addresses. func scanRouteRow( rows *sql.Rows, addresses map[uint]string, ) (routetable.Change, error) { var ( change routetable.Change prefix string peers []byte kept [keptRoutesPerRow]keptRouteColumns ) err := rows.Scan(&prefix, &change.OriginASN, &peers, &kept[0].peer, &kept[0].asPath, &kept[0].time, &kept[1].peer, &kept[1].asPath, &kept[1].time) if err != nil { return change, fmt.Errorf("failed to read a route: %w", err) } change.Prefix, err = netip.ParsePrefix(prefix) if err != nil { return change, fmt.Errorf("failed to read route %s: %w", prefix, err) } change.Peers = new(bitset.BitSet) err = change.Peers.UnmarshalBinary(peers) if err != nil { return change, fmt.Errorf("failed to read the peers of route %s: %w", prefix, err) } for _, columns := range kept { if !columns.peer.Valid { continue } address, known := addresses[columns.peer.V] if !known { return change, fmt.Errorf("%w: number %d in route %s", errUnknownRoutePeer, columns.peer.V, prefix) } r := routetable.Route{ Peer: address, Time: time.Unix(columns.time.Int64, 0).UTC(), } err := json.Unmarshal([]byte(columns.asPath.String), &r.ASPath) if err != nil { return change, fmt.Errorf("failed to read an AS path of route %s: %w", prefix, err) } change.Routes = append(change.Routes, r) } return change, nil }