mirror of
https://github.com/MeshCore-Beacon/beacon-server.git
synced 2026-10-16 07:19:58 +00:00
193 lines
7.8 KiB
Go
193 lines
7.8 KiB
Go
// Copyright 2026 Beacon Contributors
|
|
// SPDX-License-Identifier: AGPL-3.0-or-later
|
|
|
|
package db
|
|
|
|
import (
|
|
"context"
|
|
"encoding/json"
|
|
"fmt"
|
|
"net/http/httptest"
|
|
"os"
|
|
"slices"
|
|
"testing"
|
|
"time"
|
|
|
|
sqlc "github.com/MeshCore-Beacon/beacon-server/db/sqlc"
|
|
"github.com/MeshCore-Beacon/beacon-server/internal/api"
|
|
"github.com/MeshCore-Beacon/beacon-server/internal/api/handlers"
|
|
"github.com/jackc/pgx/v5"
|
|
meshcore "github.com/meshcore-go/meshcore-go"
|
|
)
|
|
|
|
func TestPathsPostgres(t *testing.T) {
|
|
dsn := os.Getenv("BEACON_TEST_POSTGRES_DSN")
|
|
if dsn == "" {
|
|
t.Skip("set BEACON_TEST_POSTGRES_DSN for PostgreSQL regression")
|
|
}
|
|
ctx, cancel := context.WithTimeout(context.Background(), 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())
|
|
applyBaseline(t, ctx, tx)
|
|
_, err = tx.Exec(ctx, `
|
|
SET LOCAL TIME ZONE 'America/Vancouver';
|
|
CREATE TEMP TABLE fixture(heard_at timestamptz NOT NULL,iata char(3) NOT NULL,payload_type smallint,path_length_byte smallint NOT NULL,hash_size smallint NOT NULL,hop_count smallint NOT NULL,path_bytes bytea);
|
|
INSERT INTO fixture
|
|
SELECT CASE id WHEN 1 THEN '2026-01-01 00:30+00'::timestamptz WHEN 17 THEN '2026-01-01 04:00+00'::timestamptz WHEN 18 THEN '2026-01-01 04:30+00'::timestamptz WHEN 20 THEN '2026-01-01 03:00+00'::timestamptz
|
|
ELSE '2026-01-01 00:30+00'::timestamptz+(id/10)*interval '2 hour'+(id%10)*interval '1 minute' END,
|
|
iata,payload,raw,width,hops,CASE WHEN size IS NULL THEN NULL ELSE decode(repeat('ab',size),'hex') END
|
|
FROM (VALUES
|
|
(1,'YVR',4,2,1,2,2),(2,'YVR',4,67,2,3,6),(3,'YYJ',5,130,3,2,6),
|
|
(4,'YVR',4,64,2,0,NULL),(5,'YVR',9,3,1,3,3),(6,'YYJ',NULL,2,1,2,2),
|
|
(7,'YVR',4,3,1,3,2),(8,'YVR',4,194,4,2,8),(9,'YVR',4,96,2,32,64),
|
|
(10,'YVR',4,97,2,33,66),(11,'YVR',4,64,1,0,0),(12,'YVR',4,0,1,1,1),
|
|
(13,'YYZ',5,0,1,0,0),(14,'YVR',16,1,1,1,1),(15,'YVR',4,63,1,63,63),
|
|
(16,'YVR',4,149,3,21,63),(17,'YVR',4,1,1,1,1),(18,'YVR',4,1,1,1,1),
|
|
(19,'YYJ',9,3,1,3,3),(20,'YVR',4,64,2,0,0)
|
|
) v(id,iata,payload,raw,width,hops,size);
|
|
ALTER TABLE fixture ADD COLUMN n serial;
|
|
INSERT INTO observers (id,public_key) VALUES ('00000000-0000-0000-0000-000000000001','\x01');
|
|
INSERT INTO packets (packet_hash,payload_type,payload_version,route_type,raw_payload,raw_header,first_heard_at,last_heard_at)
|
|
SELECT int4send(n),4,0,1,'\x00','\x00',heard_at,heard_at FROM fixture;
|
|
INSERT INTO packet_observations (packet_hash,observer_id,heard_at,iata,payload_type,path_length_byte,hash_size,hop_count,path_bytes)
|
|
SELECT int4send(n),'00000000-0000-0000-0000-000000000001',heard_at,iata,payload_type,path_length_byte,hash_size,hop_count,path_bytes FROM fixture;`)
|
|
if err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
since := time.Now().UTC().Truncate(24 * time.Hour).Add(-48 * time.Hour)
|
|
if _, err := tx.Exec(ctx, "UPDATE packet_observations SET heard_at=to_timestamp(extract(epoch FROM $1::timestamptz)+extract(epoch FROM heard_at)-extract(epoch FROM '2026-01-01 00:00+00'::timestamptz))", since); err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
rollTxHours(t, ctx, tx)
|
|
store := &Store{q: sqlc.New(tx)}
|
|
until := since.Add(4 * time.Hour)
|
|
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)
|
|
}
|
|
for _, tc := range []struct {
|
|
name string
|
|
iatas []string
|
|
counts [5]int64
|
|
}{
|
|
{"global", nil, [5]int64{18, 6, 3, 2, 7}}, {"empty filter", []string{}, [5]int64{18, 6, 3, 2, 7}},
|
|
{"YVR", []string{"YVR"}, [5]int64{14, 5, 2, 1, 6}}, {"overlap", []string{"YVR", "YYJ", "YVR"}, [5]int64{17, 6, 2, 2, 7}},
|
|
{"unknown", []string{"ZZZ"}, [5]int64{}}, {"empty region", []string{""}, [5]int64{}},
|
|
} {
|
|
t.Run(mode+"/"+tc.name, func(t *testing.T) {
|
|
got, err := store.GetPathStats(ctx, since, until, tc.iatas)
|
|
if err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
if [5]int64{got.Receptions, got.Hashed, got.Empty, got.Trace, got.Unclassified} != tc.counts {
|
|
t.Fatalf("counts: %+v", got)
|
|
}
|
|
var widths, lengths, hours int64
|
|
for _, b := range got.HashWidths {
|
|
widths += b.Receptions
|
|
}
|
|
for _, b := range got.PathLengths {
|
|
lengths += b.Receptions
|
|
}
|
|
for i, h := range got.Hourly {
|
|
hours += h.Receptions
|
|
if h.Receptions != h.OneByte+h.TwoByte+h.ThreeByte+h.Empty+h.Trace+h.Unclassified || h.Hour%3600000 != 0 || (i > 0 && h.Hour <= got.Hourly[i-1].Hour) {
|
|
t.Fatalf("hour: %+v", h)
|
|
}
|
|
}
|
|
if widths != got.Hashed || lengths != got.Hashed+got.Empty || hours != got.Receptions {
|
|
t.Fatal("partitions do not reconcile")
|
|
}
|
|
if tc.name == "global" && (got.HashWidths[0].Receptions != 2 || got.HashWidths[1].Receptions != 2 || got.HashWidths[2].Receptions != 2 || got.PathLengths[0].Entries != 0 || got.PathLengths[0].Receptions != 3 || len(got.Hourly) != 3) {
|
|
t.Fatalf("empty/trace widths or missing hour: %+v", got)
|
|
}
|
|
})
|
|
}
|
|
}
|
|
// The longest routed path per hour; hours with only empty, TRACE or invalid paths report 0.
|
|
for _, tc := range []struct {
|
|
iatas []string
|
|
want []int32
|
|
}{
|
|
{nil, []int32{32, 63, 0}},
|
|
{[]string{"YVR", "YYJ"}, []int32{32, 63, 0}},
|
|
{[]string{"YYJ"}, []int32{2, 0}},
|
|
{[]string{"YYZ"}, []int32{0}},
|
|
} {
|
|
got, err := store.GetPathStats(ctx, since, until, tc.iatas)
|
|
if err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
var max []int32
|
|
for _, h := range got.Hourly {
|
|
max = append(max, h.MaxEntries)
|
|
}
|
|
if !slices.Equal(max, tc.want) {
|
|
t.Errorf("maxEntries for %v = %v, want %v", tc.iatas, max, tc.want)
|
|
}
|
|
}
|
|
w := httptest.NewRecorder()
|
|
handlers.StatsRouter(store, handlers.StatsOptions{}).ServeHTTP(w, httptest.NewRequest("GET", fmt.Sprintf("/paths?since=%d&until=%d&iatas=YVR", since.UnixMilli()+123, until.UnixMilli()+123), nil).WithContext(ctx))
|
|
var response api.PathStats
|
|
if err := json.Unmarshal(w.Body.Bytes(), &response); err != nil || w.Code != 200 || response.Receptions != 14 || response.Hashed != 5 {
|
|
t.Fatalf("HTTP %d: %s (%v)", w.Code, w.Body.String(), err)
|
|
}
|
|
if _, err = tx.Exec(ctx, "TRUNCATE packet_observations"); err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
if _, err = tx.Exec(ctx, `INSERT INTO packets (packet_hash,payload_type,payload_version,route_type,raw_payload,raw_header,first_heard_at,last_heard_at)
|
|
SELECT int4send(1000+i),4,0,1,'\x00','\x00',$1,$1 FROM generate_series(0,255)i`, since); err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
if _, err = tx.Exec(ctx, `INSERT INTO packet_observations (packet_hash,observer_id,heard_at,iata,payload_type,path_length_byte,hash_size,hop_count,path_bytes)
|
|
SELECT int4send(1000+i),'00000000-0000-0000-0000-000000000001',$1::timestamptz+interval '1 hour','YVR',4,i,(i>>6)+1,i&63,decode(repeat('ab',((i>>6)+1)*(i&63)),'hex') FROM generate_series(0,255)i;`, since); err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
stale, err := store.GetPathStats(ctx, since, until, nil)
|
|
if err != nil || stale.Receptions != 18 {
|
|
t.Fatalf("snapshot lost: %+v %v", stale, err)
|
|
}
|
|
rollTxHours(t, ctx, tx)
|
|
got, err := store.GetPathStats(ctx, since, until, nil)
|
|
if err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
var hashed, empty, unknown int64
|
|
var widths [3]int64
|
|
for raw := 0; raw < 256; raw++ {
|
|
if !meshcore.IsValidPathLen(uint8(raw)) {
|
|
unknown++
|
|
} else if raw&63 == 0 {
|
|
empty++
|
|
} else {
|
|
hashed++
|
|
widths[raw>>6]++
|
|
}
|
|
}
|
|
if got.Hashed != hashed || got.Empty != empty || got.Unclassified != unknown || got.Receptions != 256 {
|
|
t.Fatalf("decoder agreement: %+v", got)
|
|
}
|
|
for i, count := range widths {
|
|
if got.HashWidths[i].Receptions != count {
|
|
t.Fatal("decoder width disagreement")
|
|
}
|
|
}
|
|
if _, err = tx.Exec(ctx, "TRUNCATE packet_observations"); err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
rollTxHours(t, ctx, tx)
|
|
got, err = store.GetPathStats(ctx, since, until, nil)
|
|
if err != nil || got.Receptions != 0 || got.Hourly == nil || got.PathLengths == nil || len(got.HashWidths) != 3 {
|
|
t.Fatalf("empty: %+v %v", got, err)
|
|
}
|
|
}
|