beacon-server/db/clock_drift_integration_test.go

// Copyright 2026 Beacon Contributors
// SPDX-License-Identifier: AGPL-3.0-or-later

package db

import (
	"context"
	_ "embed"
	"encoding/json"
	"fmt"
	"os"
	"slices"
	"strings"
	"testing"
	"time"

	sqlc "github.com/MeshCore-Beacon/beacon-server/db/sqlc"
	"github.com/MeshCore-Beacon/beacon-server/internal/api"
	"github.com/jackc/pgx/v5"
)

//go:embed queries/queries.sql
var clockTestQueries string

// Fixture writes are confined to temporary tables in a rolled-back transaction.
func TestClockDriftPagePostgres(t *testing.T) {
	dsn := os.Getenv("BEACON_TEST_POSTGRES_DSN")
	if dsn == "" {
		t.Skip("set BEACON_TEST_POSTGRES_DSN for the PostgreSQL regression test")
	}
	ctx, cancel := context.WithTimeout(context.Background(), 2*time.Minute)
	defer cancel()
	conn, err := pgx.Connect(ctx, dsn)
	if err != nil {
		t.Fatal(err)
	}
	defer conn.Close(context.Background())
	tx, err := conn.Begin(ctx)
	if err != nil {
		t.Fatal(err)
	}
	defer tx.Rollback(context.Background())
	for _, table := range []string{"nodes", "node_iatas"} {
		if _, err := tx.Exec(ctx, "CREATE TEMP TABLE "+table+" (LIKE public."+table+" INCLUDING ALL) ON COMMIT DROP"); err != nil {
			t.Fatal(err)
		}
	}
	_, err = tx.Exec(ctx, `
INSERT INTO nodes (id,public_key,node_type,name,device_clock_drift_seconds,last_advert_at)
SELECT ('00000000-0000-0000-0000-'||lpad(i::text,12,'0'))::uuid,decode(lpad(to_hex(i),64,'0'),'hex'),kind,
 CASE WHEN i=2 THEN NULL ELSE 'node-'||i END,drift,CASE WHEN i=2 THEN NULL ELSE '2026-01-01 00:00:00+00'::timestamptz END
FROM (VALUES (1,2,120),(2,3,-240),(3,1,99999),(4,4,-99999),(5,2,NULL),(6,2,60),(7,3,-60),(8,2,180)) v(i,kind,drift);
INSERT INTO node_iatas (node_id,iata,last_heard) VALUES
 ('00000000-0000-0000-0000-000000000001','YVR','2026-01-01 00:00:00+00'),
 ('00000000-0000-0000-0000-000000000001','YYJ','2026-01-01 01:00:00+00'),
 ('00000000-0000-0000-0000-000000000002','YYJ','2026-01-01 00:00:00+00');`)
	if err != nil {
		t.Fatal(err)
	}
	q := sqlc.New(tx)
	anchor := time.Date(2026, 1, 1, 0, 0, 0, 0, time.UTC)
	for _, tc := range []struct {
		name             string
		iatas            []string
		threshold, limit int32
		want             []int
	}{
		{"global", nil, 60, 10, []int{2, 8, 1}},
		{"empty region array", []string{}, 60, 10, []int{2, 8, 1}},
		{"page limit", nil, 60, 2, []int{2, 8}},
		{"YVR", []string{"YVR"}, 60, 10, []int{1}},
		{"YYJ", []string{"YYJ"}, 60, 10, []int{2, 1}},
		{"multiple regions", []string{"YVR", "YYJ"}, 60, 10, []int{2, 1}},
		{"duplicate region", []string{"YYJ", "YYJ"}, 60, 10, []int{2, 1}},
		{"unknown region", []string{"ZZZ"}, 60, 10, nil},
		{"strict threshold", nil, 180, 10, []int{2}},
		{"boundary", nil, 240, 10, nil},
		{"positive and negative boundary", nil, 0, 10, []int{2, 8, 1, 6, 7}},
	} {
		t.Run(tc.name, func(t *testing.T) {
			rows, err := q.GetStatsClockDrift(ctx, sqlc.GetStatsClockDriftParams{Column1: tc.threshold, Column2: tc.iatas, Limit: tc.limit})
			if err != nil {
				t.Fatal(err)
			}
			if len(rows) != len(tc.want) {
				t.Fatalf("got %d rows, want %d", len(rows), len(tc.want))
			}
			previous := int64(1 << 62)
			seen := map[int]bool{}
			for _, row := range rows {
				var number int
				if _, err := fmt.Sscanf(row.ID.String(), "00000000-0000-0000-0000-%d", &number); err != nil {
					t.Fatal(err)
				}
				if !slices.Contains(tc.want, number) || seen[number] {
					t.Fatalf("unexpected/duplicate node %d", number)
				}
				seen[number] = true
				if row.DeviceClockDriftSeconds == nil {
					t.Fatal("null drift returned")
				}
				wantDrift := map[int]int32{1: 120, 2: -240, 6: 60, 7: -60, 8: 180}[number]
				if *row.DeviceClockDriftSeconds != wantDrift || (row.NodeType != 2 && row.NodeType != 3) {
					t.Fatal("drift sign or role changed")
				}
				magnitude := int64(wantDrift)
				if magnitude < 0 {
					magnitude = -magnitude
				}
				if magnitude > previous {
					t.Fatal("not ordered worst first")
				}
				previous = magnitude
				if number == 2 {
					if row.Name != nil || row.LastAdvertAt.Valid {
						t.Fatal("nullable node fields changed")
					}
				} else if row.Name == nil || *row.Name != fmt.Sprintf("node-%d", number) || !row.LastAdvertAt.Valid || !row.LastAdvertAt.Time.Equal(anchor) {
					t.Fatal("node fields changed")
				}
				var got []api.NodeIATA
				if len(row.Iatas) > 0 {
					if err := json.Unmarshal(row.Iatas, &got); err != nil {
						t.Fatal(err)
					}
				}
				var want []api.NodeIATA
				if number == 1 {
					want = []api.NodeIATA{{IATA: "YYJ", LastHeard: anchor.Add(time.Hour).UnixMilli()}, {IATA: "YVR", LastHeard: anchor.UnixMilli()}}
				}
				if number == 2 {
					want = []api.NodeIATA{{IATA: "YYJ", LastHeard: anchor.UnixMilli()}}
				}
				if !slices.Equal(got, want) {
					t.Fatalf("node %d memberships/order: got %v, want %v", number, got, want)
				}
			}
		})
	}
	// Use a separate large population for the work-budget measurement.
	_, err = tx.Exec(ctx, `
TRUNCATE pg_temp.node_iatas,pg_temp.nodes;
INSERT INTO nodes (id,public_key,node_type,name,device_clock_drift_seconds,last_advert_at)
SELECT md5(i::text)::uuid,decode(lpad(to_hex(i),64,'0'),'hex'),2,'clock-'||i,3600+i,'2026-01-01 00:00:00+00'
FROM generate_series(1,20000) i;
INSERT INTO node_iatas (node_id,iata,last_heard)
SELECT md5(i::text)::uuid,iata,'2026-01-01 00:00:00+00'::timestamptz-i*interval '1 second'
FROM generate_series(1,20000) i CROSS JOIN (VALUES ('YVR'),('YYJ')) regions(iata);
ANALYZE nodes; ANALYZE node_iatas;`)
	if err != nil {
		t.Fatal(err)
	}
	queries := strings.ReplaceAll(clockTestQueries, "\r\n", "\n")
	query := strings.Split(strings.Split(queries, "-- name: GetStatsClockDrift :many\n")[1], "\n-- name:")[0]
	if _, err := tx.Exec(ctx, "PREPARE clock_page AS "+query); err != nil {
		t.Fatal(err)
	}
	for _, mode := range []string{"force_custom_plan", "force_generic_plan"} {
		if _, err := tx.Exec(ctx, "SET LOCAL plan_cache_mode = "+mode); err != nil {
			t.Fatal(err)
		}
		var raw []byte
		if err := tx.QueryRow(ctx, "EXPLAIN (ANALYZE, BUFFERS, FORMAT JSON) EXECUTE clock_page(60,NULL,50)").Scan(&raw); err != nil {
			t.Fatal(err)
		}
		var plans []struct {
			Plan          map[string]any
			ExecutionTime float64 `json:"Execution Time"`
		}
		if err := json.Unmarshal(raw, &plans); err != nil {
			t.Fatal(err)
		}
		var memberships float64
		var walk func(map[string]any)
		walk = func(p map[string]any) {
			if p["Relation Name"] == "node_iatas" {
				memberships += p["Actual Rows"].(float64) * p["Actual Loops"].(float64)
			}
			if children, ok := p["Plans"].([]any); ok {
				for _, child := range children {
					walk(child.(map[string]any))
				}
			}
		}
		walk(plans[0].Plan)
		t.Logf("20,000 nodes / 50-row clock page %s: %.3f ms, %.0f membership rows processed", mode, plans[0].ExecutionTime, memberships)
		if memberships > 150 {
			t.Errorf("clock page processed %.0f membership rows; budget 150", memberships)
		}
	}
}