routewatch/internal/database/originchanges.go

package database

import (
	"context"
	"fmt"
	"net/netip"
	"time"

	"git.eeqj.de/sneak/routewatch/internal/routetable"
)

// maxOriginChangesPerPrefix is how many origin changes the prune keeps for
// each prefix: the newest ones. A prefix with more has its older rows deleted.
const maxOriginChangesPerPrefix = 500

// originChangesRetention is how long the prune keeps an origin change: rows
// older than this are deleted, unless they are among the prefix's newest
// maxOriginChangesPerPrefix.
const originChangesRetention = 72 * time.Hour

// The statements of the origin_changes table (schema.sql).
const (
	insertOriginChangeSQL = `INSERT INTO origin_changes
		(prefix, origin_asn, announced, peer_count, time)
		VALUES (?, ?, ?, ?, ?)`
	originChangesSQL = `SELECT prefix, origin_asn, announced, peer_count, time
		FROM origin_changes WHERE prefix = ? ORDER BY id DESC LIMIT ?`
	prunePrefixesSQL = `SELECT prefix FROM origin_changes
		GROUP BY prefix HAVING COUNT(*) > ?`
	pruneOriginChangesSQL = `DELETE FROM origin_changes
		WHERE prefix = ? AND time < ? AND id < (
			SELECT id FROM origin_changes WHERE prefix = ?
			ORDER BY id DESC LIMIT 1 OFFSET ?)`
)

// WriteOriginChanges writes, in one transaction, the origin changes the
// routing table handed over. When it fails, it writes nothing; the caller then
// writes them again.
func (d *Database) WriteOriginChanges(changes []routetable.OriginChange) error {
	if len(changes) == 0 {
		return nil
	}

	d.lock("WriteOriginChanges")
	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()

	stmt, err := tx.PrepareContext(ctx, insertOriginChangeSQL)
	if err != nil {
		return fmt.Errorf("failed to prepare origin change insert: %w", err)
	}
	defer func() { _ = stmt.Close() }()

	for _, change := range changes {
		_, err := stmt.ExecContext(ctx, change.Prefix.String(),
			change.OriginASN, change.Announced, change.PeerCount,
			change.Time.Unix())
		if err != nil {
			return fmt.Errorf("failed to write origin change %s: %w",
				change.Prefix, err)
		}
	}

	err = tx.Commit()
	if err != nil {
		return fmt.Errorf("failed to commit transaction: %w", err)
	}

	return nil
}

// GetOriginChanges returns at most limit of prefix's origin changes, newest
// first.
func (d *Database) GetOriginChanges(
	ctx context.Context, prefix netip.Prefix, limit int,
) ([]routetable.OriginChange, error) {
	rows, err := d.db.QueryContext(ctx, originChangesSQL,
		prefix.Masked().String(), limit)
	if err != nil {
		return nil, fmt.Errorf("failed to read origin changes: %w", err)
	}
	defer func() { _ = rows.Close() }()

	var changes []routetable.OriginChange

	for rows.Next() {
		var (
			change     routetable.OriginChange
			prefixText string
			timeSec    int64
		)

		err := rows.Scan(&prefixText, &change.OriginASN, &change.Announced,
			&change.PeerCount, &timeSec)
		if err != nil {
			return nil, fmt.Errorf("failed to read an origin change: %w", err)
		}

		change.Prefix, err = netip.ParsePrefix(prefixText)
		if err != nil {
			return nil, fmt.Errorf("failed to read origin change %s: %w",
				prefixText, err)
		}

		change.Time = time.Unix(timeSec, 0).UTC()
		changes = append(changes, change)
	}

	err = rows.Err()
	if err != nil {
		return nil, fmt.Errorf("failed to read origin changes: %w", err)
	}

	return changes, nil
}

// PruneOriginChanges deletes, for each prefix with more than
// maxOriginChangesPerPrefix rows, the rows older than originChangesRetention
// that are not among its newest maxOriginChangesPerPrefix. It finds those
// prefixes on a reading connection and deletes one prefix per write, so no
// write holds the lock for long.
func (d *Database) PruneOriginChanges(ctx context.Context) error {
	rows, err := d.db.QueryContext(ctx, prunePrefixesSQL,
		maxOriginChangesPerPrefix)
	if err != nil {
		return fmt.Errorf("failed to find prefixes to prune: %w", err)
	}
	defer func() { _ = rows.Close() }()

	var prefixes []string

	for rows.Next() {
		var prefix string

		err := rows.Scan(&prefix)
		if err != nil {
			return fmt.Errorf("failed to read a prefix to prune: %w", err)
		}

		prefixes = append(prefixes, prefix)
	}

	err = rows.Err()
	if err != nil {
		return fmt.Errorf("failed to find prefixes to prune: %w", err)
	}

	cutoff := time.Now().Add(-originChangesRetention).Unix()

	for _, prefix := range prefixes {
		d.lock("PruneOriginChanges")

		_, err := d.writer.ExecContext(ctx, pruneOriginChangesSQL,
			prefix, cutoff, prefix, maxOriginChangesPerPrefix-1)

		d.unlock()

		if err != nil {
			return fmt.Errorf("failed to prune origin changes of %s: %w",
				prefix, err)
		}
	}

	return nil
}