routewatch/internal/database/database_test.go

package database

import (
	"context"
	"database/sql"
	"errors"
	"net/netip"
	"sync"
	"testing"
	"time"

	"git.eeqj.de/sneak/routewatch/internal/config"
	"git.eeqj.de/sneak/routewatch/internal/logger"
	"github.com/google/uuid"
)

// tempStoreMemory is the PRAGMA temp_store value meaning "hold temp B-trees in
// memory"; the DSN change must leave temp_store below this so they spill to disk.
const tempStoreMemory = 2

// heldConnections is how many pooled connections the pragma test holds open at
// once so each is a distinct SQLite connection that parsed the DSN.
const heldConnections = 5

// testPeerIP is the peer every route in the IP lookup test is learned from.
const testPeerIP = "192.0.2.254"

// Parameters for the checkpoint-contention regression test.
const (
	// contentionIterations is how many batch writes race the checkpoint loop.
	contentionIterations = 400
	// contendedASNCount is the small set of ASNs the batches reuse, so most
	// batches update existing rows and exercise the read-before-write path.
	contendedASNCount = 16
	// asnSecondBand offsets a second ASN per batch so each batch writes more
	// than one row.
	asnSecondBand = 100
)

// TestGetIPInfoFindsMostSpecificLiveRoute stores nested live prefixes for both
// families and checks that a lookup returns the most specific one covering the
// address, ErrNoRoute when none covers it, and the next less specific prefix
// once the only route of the most specific one is withdrawn.
func TestGetIPInfoFindsMostSpecificLiveRoute(t *testing.T) {
	cfg := &config.Config{StateDir: t.TempDir()}

	db, err := New(cfg, logger.New())
	if err != nil {
		t.Fatalf("failed to create database: %v", err)
	}
	defer func() { _ = db.Close() }()

	// Nested live prefixes, each originated by its own AS.
	origins := map[string]int{
		"10.0.0.0/8":        64500,
		"10.1.0.0/16":       64501,
		"10.1.2.0/24":       64502,
		"2001:db8::/32":     64500,
		"2001:db8:1::/48":   64501,
		"2001:db8:1:2::/64": 64502,
	}
	ts := time.Date(2026, 1, 2, 3, 4, 5, 0, time.UTC)
	routes := make([]*LiveRoute, 0, len(origins))
	for prefix, asn := range origins {
		routes = append(routes, &LiveRoute{
			ID:          uuid.New(),
			Prefix:      prefix,
			MaskLength:  netip.MustParsePrefix(prefix).Bits(),
			IPVersion:   detectIPVersion(prefix),
			OriginASN:   asn,
			PeerIP:      testPeerIP,
			ASPath:      []int{asn},
			NextHop:     testPeerIP,
			LastUpdated: ts,
		})
	}
	if err := db.UpsertLiveRouteBatch(routes); err != nil {
		t.Fatalf("UpsertLiveRouteBatch: %v", err)
	}

	// lookup checks that ip resolves to the live prefix want, or to ErrNoRoute
	// when want is empty.
	lookup := func(ip, want string) {
		t.Helper()

		info, err := db.GetIPInfo(ip)
		if want == "" {
			if !errors.Is(err, ErrNoRoute) {
				t.Errorf("GetIPInfo(%s) = %+v, %v; want ErrNoRoute", ip, info, err)
			}

			return
		}
		if err != nil {
			t.Errorf("GetIPInfo(%s): %v", ip, err)

			return
		}
		if info.Netblock != want || info.MaskLength != netip.MustParsePrefix(want).Bits() ||
			info.ASN != origins[want] {
			t.Errorf("GetIPInfo(%s) = %s (mask %d) AS%d, want %s AS%d",
				ip, info.Netblock, info.MaskLength, info.ASN, want, origins[want])
		}
	}

	lookup("10.1.2.3", "10.1.2.0/24")
	lookup("::ffff:10.1.2.3", "10.1.2.0/24")
	lookup("10.1.3.4", "10.1.0.0/16")
	lookup("10.2.0.1", "10.0.0.0/8")
	lookup("192.0.2.1", "")
	lookup("2001:db8:1:2::3", "2001:db8:1:2::/64")
	lookup("2001:db8:1:3::4", "2001:db8:1::/48")
	lookup("2001:db8:2::1", "2001:db8::/32")
	lookup("2001:db9::1", "")

	err = db.DeleteLiveRouteBatch([]LiveRouteDeletion{
		{Prefix: "10.1.2.0/24", OriginASN: 64502, PeerIP: testPeerIP, IPVersion: ipVersionV4},
		{Prefix: "2001:db8:1:2::/64", OriginASN: 64502, PeerIP: testPeerIP, IPVersion: ipVersionV6},
	})
	if err != nil {
		t.Fatalf("DeleteLiveRouteBatch: %v", err)
	}

	lookup("10.1.2.3", "10.1.0.0/16")
	lookup("2001:db8:1:2::3", "2001:db8:1::/48")
}

// TestConnectionPoolPragmas holds several pooled connections open at once and
// checks each one carries the per-connection settings from the DSN, plus the
// process-wide hard heap limit.
func TestConnectionPoolPragmas(t *testing.T) {
	cfg := &config.Config{StateDir: t.TempDir()}

	db, err := New(cfg, logger.New())
	if err != nil {
		t.Fatalf("failed to create database: %v", err)
	}
	defer func() { _ = db.Close() }()

	ctx := context.Background()

	// Hold distinct connections open simultaneously so the pool must open a new
	// one (each parsing the DSN) rather than hand back the same connection.
	conns := make([]*sql.Conn, 0, heldConnections)
	defer func() {
		for _, c := range conns {
			_ = c.Close()
		}
	}()

	for i := 0; i < heldConnections; i++ {
		c, err := db.db.Conn(ctx)
		if err != nil {
			t.Fatalf("failed to open connection %d: %v", i, err)
		}
		conns = append(conns, c)
	}

	for i, c := range conns {
		var cacheSize int
		if err := c.QueryRowContext(ctx, "PRAGMA cache_size").Scan(&cacheSize); err != nil {
			t.Fatalf("conn %d: failed to read cache_size: %v", i, err)
		}
		if cacheSize != sqliteCacheSizeKiB {
			t.Errorf("conn %d: cache_size = %d, want %d", i, cacheSize, sqliteCacheSizeKiB)
		}

		var busyTimeout int
		if err := c.QueryRowContext(ctx, "PRAGMA busy_timeout").Scan(&busyTimeout); err != nil {
			t.Fatalf("conn %d: failed to read busy_timeout: %v", i, err)
		}
		if busyTimeout != sqliteBusyTimeoutMs {
			t.Errorf("conn %d: busy_timeout = %d, want %d", i, busyTimeout, sqliteBusyTimeoutMs)
		}

		var tempStore int
		if err := c.QueryRowContext(ctx, "PRAGMA temp_store").Scan(&tempStore); err != nil {
			t.Fatalf("conn %d: failed to read temp_store: %v", i, err)
		}
		if tempStore == tempStoreMemory {
			t.Errorf("conn %d: temp_store = %d, want anything but %d (MEMORY)", i, tempStore, tempStoreMemory)
		}

		var hardHeapLimit int64
		if err := c.QueryRowContext(ctx, "PRAGMA hard_heap_limit").Scan(&hardHeapLimit); err != nil {
			t.Fatalf("conn %d: failed to read hard_heap_limit: %v", i, err)
		}
		if hardHeapLimit != sqliteHardHeapLimitBytes {
			t.Errorf("conn %d: hard_heap_limit = %d, want %d", i, hardHeapLimit, sqliteHardHeapLimitBytes)
		}
	}
}

// TestBatchWriteDuringCheckpoint reproduces issue #25. A batch write reads
// (SELECT) before it writes (INSERT/UPDATE). Under the default deferred locking
// the transaction begins as a reader and, when the maintainer's WAL checkpoint
// holds the write lock, its upgrade to writer fails immediately with "database
// is locked" without honouring busy_timeout, dropping the batch. With
// _txlock=immediate the transaction takes the write lock at BEGIN and waits, so
// no batch is dropped. The checkpoint runs without the Database mutex, exactly
// as the background maintainer does in production.
func TestBatchWriteDuringCheckpoint(t *testing.T) {
	cfg := &config.Config{StateDir: t.TempDir()}

	db, err := New(cfg, logger.New())
	if err != nil {
		t.Fatalf("failed to create database: %v", err)
	}
	defer func() { _ = db.Close() }()

	ctx, cancel := context.WithCancel(context.Background())

	var wg sync.WaitGroup
	wg.Add(1)
	go func() {
		defer wg.Done()
		for {
			select {
			case <-ctx.Done():
				return
			default:
				_ = db.Checkpoint(ctx) // errors are the checkpoint's own to absorb
			}
		}
	}()

	ts := time.Now().UTC()
	for i := 0; i < contentionIterations; i++ {
		asns := map[int]time.Time{
			i % contendedASNCount:                   ts,
			(i % contendedASNCount) + asnSecondBand: ts,
		}
		if err := db.GetOrCreateASNBatch(asns); err != nil {
			cancel()
			wg.Wait()
			t.Fatalf("batch write failed under checkpoint contention: %v", err)
		}
	}

	cancel()
	wg.Wait()
}