Merge remote-tracking branch 'origin/dev' into codex/post20-base-20261002

This commit is contained in:
n30nex committed 2026-10-02 16:34:31 -04:00
commit ced19727ea
26 files changed
+485 -44

No files matched your search

+18
View File
@@ -547,6 +547,24 @@ CREATE TABLE analytics_hourly_scope_sets (
PRIMARY KEY (hour, iatas, scope_id)
);
-- Observers that heard a scoped packet, per IATA; distinct counts span any IATA filter.
CREATE TABLE analytics_hourly_scope_observers (
hour timestamp with time zone NOT NULL,
iata character(3) NOT NULL,
scope_id integer NOT NULL,
observer_id uuid NOT NULL,
PRIMARY KEY (hour, scope_id, iata, observer_id)
);
-- Nodes whose ADVERTs were heard in a scope, per IATA.
CREATE TABLE analytics_hourly_scope_nodes (
hour timestamp with time zone NOT NULL,
iata character(3) NOT NULL,
scope_id integer NOT NULL,
origin_pubkey bytea NOT NULL,
PRIMARY KEY (hour, scope_id, iata, origin_pubkey)
);
CREATE TABLE analytics_rollup_hours (
hour timestamp with time zone PRIMARY KEY,
status text NOT NULL CHECK (status IN ('complete', 'partial', 'missing')),
+1
View File
@@ -31,6 +31,7 @@ func (s *Store) GetPathStats(ctx context.Context, since, until time.Time, iatas
}
point := &stats.Hourly[len(stats.Hourly)-1]
point.Receptions += row.Receptions
point.MaxEntries = max(point.MaxEntries, row.MaxEntries)
switch row.Category {
case 0:
switch row.HashBytes {
+23
View File
@@ -9,6 +9,7 @@ import (
"fmt"
"net/http/httptest"
"os"
"slices"
"testing"
"time"
@@ -112,6 +113,28 @@ SELECT int4send(n),'00000000-0000-0000-0000-000000000001',heard_at,iata,payload_
})
}
}
// 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
+27
View File
@@ -9,6 +9,7 @@ import (
"time"
sqlc "github.com/MeshCore-Beacon/beacon-server/db/sqlc"
"github.com/jackc/pgx/v5/pgtype"
)
type pathBucketQuerier struct {
@@ -31,3 +32,29 @@ func TestPathStatsRejectsInvalidDatabaseBucket(t *testing.T) {
}
}
}
type pathRowsQuerier struct {
sqlc.Querier
rows []sqlc.GetPathStatsRow
}
func (q pathRowsQuerier) GetPathStats(context.Context, sqlc.GetPathStatsParams) ([]sqlc.GetPathStatsRow, error) {
return q.rows, nil
}
// An hour's maxEntries is the largest across its category/width rows, not the last one seen.
func TestPathStatsHourlyMaxEntries(t *testing.T) {
hour := pgtype.Timestamptz{Time: time.Date(2026, 1, 1, 0, 0, 0, 0, time.UTC), Valid: true}
store := &Store{q: pathRowsQuerier{rows: []sqlc.GetPathStatsRow{
{IsHourly: true, Hour: hour, Category: 0, HashBytes: 1, Receptions: 3, MaxEntries: 9},
{IsHourly: true, Hour: hour, Category: 0, HashBytes: 2, Receptions: 1, MaxEntries: 5},
{IsHourly: true, Hour: hour, Category: 1, HashBytes: 0, Receptions: 2, MaxEntries: 0},
}}}
got, err := store.GetPathStats(context.Background(), time.Time{}, time.Time{}, nil)
if err != nil {
t.Fatal(err)
}
if len(got.Hourly) != 1 || got.Hourly[0].MaxEntries != 9 {
t.Errorf("hourly %+v, want one hour with maxEntries 9", got.Hourly)
}
}
+3 -1
View File
@@ -12,7 +12,9 @@ WITH readings AS (
SELECT (grouping(hour) = 0)::boolean AS is_hourly,
COALESCE(hour, 'epoch'::timestamptz)::timestamptz AS hour,
category, hash_bytes, COALESCE(entries, 0)::integer AS entries,
sum(receptions)::bigint AS receptions
sum(receptions)::bigint AS receptions,
-- Non-routed categories store entries = 0, so an hour of only empty paths gives 0.
COALESCE(max(entries) FILTER (WHERE receptions > 0), 0)::integer AS max_entries
FROM readings
GROUP BY GROUPING SETS ((category, hash_bytes, entries), (hour, category, hash_bytes))
ORDER BY is_hourly, hour, category, hash_bytes, entries;
+36
View File
@@ -1242,6 +1242,42 @@ WHERE ($1::text = '' OR preset = $1::text)
AND (COALESCE(cardinality($2::bpchar[]), 0) = 0 OR iata = ANY($2::bpchar[]))
ORDER BY preset, iata, source_type;
-- name: GetScopeStatsHourly :many
-- GetScopeStats packet counts split by hour (same window and IATA-set filter), with the
-- distinct observers and advertising nodes active in each scope that hour. Empty hours omitted.
WITH pk AS (
SELECT s.scope_id, s.hour, SUM(s.packets)::bigint AS n
FROM analytics_hourly_scope_sets s
WHERE s.hour >= @since::timestamptz
AND (COALESCE(cardinality(@iatas::bpchar[]), 0) = 0 OR s.iatas && @iatas::bpchar[])
GROUP BY s.scope_id, s.hour
), ob AS (
SELECT o.scope_id, o.hour, COUNT(DISTINCT o.observer_id)::bigint AS n
FROM analytics_hourly_scope_observers o
WHERE o.hour >= @since::timestamptz
AND (COALESCE(cardinality(@iatas::bpchar[]), 0) = 0 OR o.iata = ANY(@iatas::bpchar[]))
GROUP BY o.scope_id, o.hour
), nd AS (
SELECT d.scope_id, d.hour, COUNT(DISTINCT d.origin_pubkey)::bigint AS n
FROM analytics_hourly_scope_nodes d
WHERE d.hour >= @since::timestamptz
AND (COALESCE(cardinality(@iatas::bpchar[]), 0) = 0 OR d.iata = ANY(@iatas::bpchar[]))
GROUP BY d.scope_id, d.hour
), k AS (
SELECT scope_id, hour FROM pk UNION SELECT scope_id, hour FROM ob UNION SELECT scope_id, hour FROM nd
)
SELECT ts.name, k.hour,
COALESCE(pk.n, 0)::bigint AS packets,
COALESCE(ob.n, 0)::bigint AS observers,
COALESCE(nd.n, 0)::bigint AS nodes
FROM k
JOIN transport_scopes ts ON ts.id = k.scope_id
LEFT JOIN pk ON pk.scope_id = k.scope_id AND pk.hour = k.hour
LEFT JOIN ob ON ob.scope_id = k.scope_id AND ob.hour = k.hour
LEFT JOIN nd ON nd.scope_id = k.scope_id AND nd.hour = k.hour
WHERE COALESCE(pk.n, 0) > 0 OR COALESCE(ob.n, 0) > 0 OR COALESCE(nd.n, 0) > 0
ORDER BY ts.name, k.hour;
-- name: GetScopeStats :many
-- Packets since the given hour come from the IATA-set rollup (counted once per hour heard).
-- Observer and node counts are current memberships; observers filter by their latest IATA.
+19 -3
View File
@@ -94,8 +94,10 @@ WITH a AS (DELETE FROM analytics_hourly_iata_observations ta WHERE ta.hour = @ho
g AS (DELETE FROM analytics_hourly_advert_hearings tg WHERE tg.hour = @hour::timestamptz),
h AS (DELETE FROM analytics_hourly_advert_sets th WHERE th.hour = @hour::timestamptz),
i AS (DELETE FROM analytics_hourly_talker_sets ti WHERE ti.hour = @hour::timestamptz),
j AS (DELETE FROM analytics_hourly_packet_sets tj WHERE tj.hour = @hour::timestamptz)
DELETE FROM analytics_hourly_scope_sets tk WHERE tk.hour = @hour::timestamptz;
j AS (DELETE FROM analytics_hourly_packet_sets tj WHERE tj.hour = @hour::timestamptz),
k AS (DELETE FROM analytics_hourly_scope_sets tk WHERE tk.hour = @hour::timestamptz),
l AS (DELETE FROM analytics_hourly_scope_observers tl WHERE tl.hour = @hour::timestamptz)
DELETE FROM analytics_hourly_scope_nodes tm WHERE tm.hour = @hour::timestamptz;
-- name: RollIATAObservations :exec
INSERT INTO analytics_hourly_iata_observations (hour, iata, observation_count)
@@ -221,6 +223,16 @@ FROM (SELECT scope_id, array_agg(DISTINCT iata ORDER BY iata) AS iatas
GROUP BY packet_hash, scope_id) p
GROUP BY iatas, scope_id;
-- name: RollScopeObservers :exec
INSERT INTO analytics_hourly_scope_observers (hour, iata, scope_id, observer_id)
SELECT DISTINCT @hour::timestamptz, iata, scope_id, observer_id
FROM rollup_obs WHERE scope_id IS NOT NULL;
-- name: RollScopeNodes :exec
INSERT INTO analytics_hourly_scope_nodes (hour, iata, scope_id, origin_pubkey)
SELECT DISTINCT @hour::timestamptz, iata, scope_id, origin_pubkey
FROM rollup_obs WHERE payload_type = 4 AND origin_pubkey IS NOT NULL AND scope_id IS NOT NULL;
-- name: RollupContentHash :one
-- Order-independent fingerprint of every family row for the hour.
SELECT md5(concat_ws('|',
@@ -234,7 +246,9 @@ SELECT md5(concat_ws('|',
(SELECT string_agg(t::text, ',' ORDER BY t::text) FROM analytics_hourly_advert_sets t WHERE t.hour = @hour::timestamptz),
(SELECT string_agg(t::text, ',' ORDER BY t::text) FROM analytics_hourly_talker_sets t WHERE t.hour = @hour::timestamptz),
(SELECT string_agg(t::text, ',' ORDER BY t::text) FROM analytics_hourly_packet_sets t WHERE t.hour = @hour::timestamptz),
(SELECT string_agg(t::text, ',' ORDER BY t::text) FROM analytics_hourly_scope_sets t WHERE t.hour = @hour::timestamptz)
(SELECT string_agg(t::text, ',' ORDER BY t::text) FROM analytics_hourly_scope_sets t WHERE t.hour = @hour::timestamptz),
(SELECT string_agg(t::text, ',' ORDER BY t::text) FROM analytics_hourly_scope_observers t WHERE t.hour = @hour::timestamptz),
(SELECT string_agg(t::text, ',' ORDER BY t::text) FROM analytics_hourly_scope_nodes t WHERE t.hour = @hour::timestamptz)
))::text AS content_hash;
-- name: FinishRollupHour :one
@@ -275,5 +289,7 @@ WITH a AS (DELETE FROM analytics_hourly_iata_observations ta WHERE ta.hour < @cu
j AS (DELETE FROM analytics_hourly_packet_sets tj WHERE tj.hour < @cutoff::timestamptz),
k AS (DELETE FROM analytics_hourly_scope_sets tk WHERE tk.hour < @cutoff::timestamptz),
l AS (DELETE FROM analytics_dirty_hours tl WHERE tl.hour < @cutoff::timestamptz),
n AS (DELETE FROM analytics_hourly_scope_observers tn WHERE tn.hour < @cutoff::timestamptz),
o AS (DELETE FROM analytics_hourly_scope_nodes tob WHERE tob.hour < @cutoff::timestamptz),
m AS (DELETE FROM analytics_rollup_hours tm WHERE tm.hour < @cutoff::timestamptz RETURNING 1)
UPDATE analytics_raw_state SET coverage = coverage + 1 WHERE EXISTS (SELECT 1 FROM m);
+2
View File
@@ -153,6 +153,8 @@ func rollHour(ctx context.Context, tx pgx.Tx, hour time.Time) (RollOutcome, bool
"talker sets": q.RollTalkerSets,
"packet sets": q.RollPacketSets,
"scope sets": q.RollScopeSets,
"scope observers": q.RollScopeObservers,
"scope nodes": q.RollScopeNodes,
} {
if err := roll(ctx, h); err != nil {
return RollSkipped, false, fmt.Errorf("roll %s: %w", name, err)
+10 -1
View File
@@ -37,7 +37,7 @@ INSERT INTO channels (id, channel_hash) VALUES (1, '\x11');
INSERT INTO packets (packet_hash, payload_type, payload_version, route_type, origin_pubkey, scope_id, raw_payload, raw_header, first_heard_at, last_heard_at)
SELECT hash, pt, rt, 0, origin, scope, '\x00', '\x00', $1::timestamptz, $1::timestamptz FROM (VALUES
('\x01'::bytea, 4, 1, '\xee'::bytea, NULL::int), -- flood advert, two IATAs
('\x02', 4, 2, '\xee', NULL), -- direct advert, one IATA
('\x02', 4, 2, '\xee', 1), -- direct advert, scoped, one IATA
('\x03', 5, 1, NULL, 1), -- scoped channel message, two IATAs
('\x04', 5, 1, NULL, 1), -- scoped channel message, one IATA
('\x05', 2, 1, NULL, NULL), -- straddles h0 and h0+1h
@@ -129,6 +129,15 @@ var rollupOracles = map[string][2]string{
`SELECT hour, iata, observer_id, observation_count, display_name, observer_type FROM analytics_hourly_observer_identity`,
`SELECT date_trunc('hour', po.heard_at, 'UTC'), po.iata, po.observer_id, count(*), o.display_name, o.observer_type
FROM packet_observations po JOIN observers o ON o.id = po.observer_id GROUP BY 1, 2, 3, 5, 6`},
"scope observers": {
`SELECT hour, iata, scope_id, observer_id FROM analytics_hourly_scope_observers`,
`SELECT DISTINCT date_trunc('hour', po.heard_at, 'UTC'), po.iata, p.scope_id, po.observer_id
FROM packet_observations po JOIN packets p ON p.packet_hash = po.packet_hash WHERE p.scope_id IS NOT NULL`},
"scope nodes": {
`SELECT hour, iata, scope_id, origin_pubkey FROM analytics_hourly_scope_nodes`,
`SELECT DISTINCT date_trunc('hour', po.heard_at, 'UTC'), po.iata, p.scope_id, p.origin_pubkey
FROM packet_observations po JOIN packets p ON p.packet_hash = po.packet_hash
WHERE po.payload_type = 4 AND p.origin_pubkey IS NOT NULL AND p.scope_id IS NOT NULL`},
"advert hearings": {
`SELECT hour, iata, origin_pubkey, observations, last_heard, name, node_type FROM analytics_hourly_advert_hearings`,
`SELECT date_trunc('hour', po.heard_at, 'UTC'), po.iata, p.origin_pubkey, count(*), max(po.heard_at), n.name, n.node_type
+30 -20
View File
@@ -31,15 +31,16 @@ FROM (VALUES (1,'#a'),(2,'#b'),(3,'#unused')) v(i,name);
INSERT INTO observers (id,public_key,last_iata)
SELECT md5(i::text)::uuid,decode(lpad(to_hex(i),2,'0'),'hex'),iata
FROM (VALUES (1,'YVR'),(2,'YYJ'),(3,'YVR'),(4,'YVR'),(5,'YYZ')) v(i,iata);
INSERT INTO packets (packet_hash,scope_id,payload_type,payload_version,route_type,raw_payload,raw_header,first_heard_at,last_heard_at)
SELECT decode(lpad(to_hex(i),2,'0'),'hex'),scope_id,4,0,1,'\x00','\x00',$1::timestamptz,$1::timestamptz
-- every packet is an advert from its own node, so per-scope nodes equal per-scope packets
INSERT INTO packets (packet_hash,scope_id,payload_type,payload_version,route_type,origin_pubkey,raw_payload,raw_header,first_heard_at,last_heard_at)
SELECT decode(lpad(to_hex(i),2,'0'),'hex'),scope_id,4,0,1,decode(lpad(to_hex(100+i),64,'0'),'hex'),'\x00','\x00',$1::timestamptz,$1::timestamptz
FROM (VALUES (1,1),(2,1),(3,1),(4,1),(5,2),(6,2),(7,2),(8,NULL)) v(i,scope_id);
INSERT INTO observer_scopes (observer_id,scope_id)
SELECT md5(i::text)::uuid,scope_id FROM (VALUES (1,1),(2,1),(4,1),(5,1),(1,2),(2,2),(3,2)) v(i,scope_id);
-- packet 4 is never heard, so it never reaches the rollup
INSERT INTO packet_observations (id,packet_hash,observer_id,iata,heard_at,path_length_byte,hash_size,hop_count)
INSERT INTO packet_observations (id,packet_hash,observer_id,iata,heard_at,path_length_byte,hash_size,hop_count,payload_type)
SELECT id,decode(lpad(to_hex(packet),2,'0'),'hex'),md5(observer::text)::uuid,iata,
$1::timestamptz+id*interval '1 second',0,1,0
$1::timestamptz+id*interval '1 second',0,1,0,4
FROM (VALUES (1,1,1,'YVR'),(2,1,4,'YVR'),(3,1,2,'YYJ'),(4,2,2,'YVR'),
(5,3,1,'YYZ'),(6,5,3,'YVR'),(7,6,1,'YYJ'),(8,7,2,'YYZ'),(9,8,1,'YVR')) v(id,packet,observer,iata);
INSERT INTO nodes (id,public_key,node_type,default_scope_id)
@@ -62,30 +63,39 @@ SELECT md5(i::text)::uuid,iata FROM (VALUES (1,'YVR'),(1,'YYJ'),(2,'YVR'),(3,'YY
r.Close(ctx)
for _, tc := range []struct {
name string
iatas []string
since time.Time
a, b [3]int64 // packets, observers, nodes
name string
iatas []string
since time.Time
a, b [3]int64 // packets, observers, nodes
ah, bh int64 // distinct observers heard carrying the scope in hour h
}{
{"global", nil, time.Time{}, [3]int64{3, 4, 4}, [3]int64{3, 3, 2}},
{"empty filter", []string{}, time.Time{}, [3]int64{3, 4, 4}, [3]int64{3, 3, 2}},
{"YVR", []string{"YVR"}, time.Time{}, [3]int64{2, 2, 2}, [3]int64{1, 2, 1}},
{"YYJ", []string{"YYJ"}, time.Time{}, [3]int64{1, 1, 1}, [3]int64{1, 1, 2}},
{"YYZ", []string{"YYZ"}, time.Time{}, [3]int64{1, 1, 1}, [3]int64{1, 0, 0}},
{"overlapping IATAs", []string{"YVR", "YYJ"}, time.Time{}, [3]int64{2, 3, 2}, [3]int64{2, 3, 2}},
{"duplicate IATAs", []string{"YYJ", "YVR", "YYJ"}, time.Time{}, [3]int64{2, 3, 2}, [3]int64{2, 3, 2}},
{"unknown IATA", []string{"ZZZ"}, time.Time{}, [3]int64{}, [3]int64{}},
{"window after the traffic", nil, h.Add(time.Hour), [3]int64{0, 4, 4}, [3]int64{0, 3, 2}},
{"global", nil, time.Time{}, [3]int64{3, 4, 4}, [3]int64{3, 3, 2}, 3, 3},
{"empty filter", []string{}, time.Time{}, [3]int64{3, 4, 4}, [3]int64{3, 3, 2}, 3, 3},
{"YVR", []string{"YVR"}, time.Time{}, [3]int64{2, 2, 2}, [3]int64{1, 2, 1}, 3, 1},
{"YYJ", []string{"YYJ"}, time.Time{}, [3]int64{1, 1, 1}, [3]int64{1, 1, 2}, 1, 1},
{"YYZ", []string{"YYZ"}, time.Time{}, [3]int64{1, 1, 1}, [3]int64{1, 0, 0}, 1, 1},
{"overlapping IATAs", []string{"YVR", "YYJ"}, time.Time{}, [3]int64{2, 3, 2}, [3]int64{2, 3, 2}, 3, 2},
{"duplicate IATAs", []string{"YYJ", "YVR", "YYJ"}, time.Time{}, [3]int64{2, 3, 2}, [3]int64{2, 3, 2}, 3, 2},
{"unknown IATA", []string{"ZZZ"}, time.Time{}, [3]int64{}, [3]int64{}, 0, 0},
{"window after the traffic", nil, h.Add(time.Hour), [3]int64{0, 4, 4}, [3]int64{0, 3, 2}, 0, 0},
} {
t.Run(tc.name, func(t *testing.T) {
rows, err := store.GetScopeStats(ctx, tc.iatas, tc.since)
if err != nil {
t.Fatal(err)
}
// All fixture traffic is in hour h; hours with no packets are omitted.
// Each packet has its own origin, so active nodes per hour equal packets per hour.
hourly := func(packets, observers int64) []api.ScopeHour {
if packets == 0 && observers == 0 {
return []api.ScopeHour{}
}
return []api.ScopeHour{{Hour: h.UnixMilli(), Packets: packets, Observers: observers, Nodes: packets}}
}
want := []api.ScopeStats{
{Name: "#a", PacketCount: tc.a[0], ObserverCount: tc.a[1], NodeCount: tc.a[2]},
{Name: "#b", PacketCount: tc.b[0], ObserverCount: tc.b[1], NodeCount: tc.b[2]},
{Name: "#unused"},
{Name: "#a", PacketCount: tc.a[0], ObserverCount: tc.a[1], NodeCount: tc.a[2], Hourly: hourly(tc.a[0], tc.ah)},
{Name: "#b", PacketCount: tc.b[0], ObserverCount: tc.b[1], NodeCount: tc.b[2], Hourly: hourly(tc.b[0], tc.bh)},
{Name: "#unused", Hourly: []api.ScopeHour{}},
}
if !reflect.DeepEqual(rows, want) {
t.Fatalf("counts = %+v, want %+v", rows, want)
+43
View File
@@ -972,6 +972,21 @@ func (mr *MockQuerierMockRecorder) GetScopeStats(ctx, arg any) *gomock.Call {
return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "GetScopeStats", reflect.TypeOf((*MockQuerier)(nil).GetScopeStats), ctx, arg)
}
// GetScopeStatsHourly mocks base method.
func (m *MockQuerier) GetScopeStatsHourly(ctx context.Context, arg db.GetScopeStatsHourlyParams) ([]db.GetScopeStatsHourlyRow, error) {
m.ctrl.T.Helper()
ret := m.ctrl.Call(m, "GetScopeStatsHourly", ctx, arg)
ret0, _ := ret[0].([]db.GetScopeStatsHourlyRow)
ret1, _ := ret[1].(error)
return ret0, ret1
}
// GetScopeStatsHourly indicates an expected call of GetScopeStatsHourly.
func (mr *MockQuerierMockRecorder) GetScopeStatsHourly(ctx, arg any) *gomock.Call {
mr.mock.ctrl.T.Helper()
return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "GetScopeStatsHourly", reflect.TypeOf((*MockQuerier)(nil).GetScopeStatsHourly), ctx, arg)
}
// GetSignalStats mocks base method.
func (m *MockQuerier) GetSignalStats(ctx context.Context, arg db.GetSignalStatsParams) ([]db.GetSignalStatsRow, error) {
m.ctrl.T.Helper()
@@ -1963,6 +1978,34 @@ func (mr *MockQuerierMockRecorder) RollPayloadBreakdown(ctx, hour any) *gomock.C
return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "RollPayloadBreakdown", reflect.TypeOf((*MockQuerier)(nil).RollPayloadBreakdown), ctx, hour)
}
// RollScopeNodes mocks base method.
func (m *MockQuerier) RollScopeNodes(ctx context.Context, hour pgtype.Timestamptz) error {
m.ctrl.T.Helper()
ret := m.ctrl.Call(m, "RollScopeNodes", ctx, hour)
ret0, _ := ret[0].(error)
return ret0
}
// RollScopeNodes indicates an expected call of RollScopeNodes.
func (mr *MockQuerierMockRecorder) RollScopeNodes(ctx, hour any) *gomock.Call {
mr.mock.ctrl.T.Helper()
return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "RollScopeNodes", reflect.TypeOf((*MockQuerier)(nil).RollScopeNodes), ctx, hour)
}
// RollScopeObservers mocks base method.
func (m *MockQuerier) RollScopeObservers(ctx context.Context, hour pgtype.Timestamptz) error {
m.ctrl.T.Helper()
ret := m.ctrl.Call(m, "RollScopeObservers", ctx, hour)
ret0, _ := ret[0].(error)
return ret0
}
// RollScopeObservers indicates an expected call of RollScopeObservers.
func (mr *MockQuerierMockRecorder) RollScopeObservers(ctx, hour any) *gomock.Call {
mr.mock.ctrl.T.Helper()
return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "RollScopeObservers", reflect.TypeOf((*MockQuerier)(nil).RollScopeObservers), ctx, hour)
}
// RollScopeSets mocks base method.
func (m *MockQuerier) RollScopeSets(ctx context.Context, hour pgtype.Timestamptz) error {
m.ctrl.T.Helper()
+14
View File
@@ -91,6 +91,20 @@ type AnalyticsHourlyPayloadBreakdown struct {
Count int64 `json:"count"`
}
type AnalyticsHourlyScopeNode struct {
Hour pgtype.Timestamptz `json:"hour"`
Iata string `json:"iata"`
ScopeID int32 `json:"scope_id"`
OriginPubkey []byte `json:"origin_pubkey"`
}
type AnalyticsHourlyScopeObserver struct {
Hour pgtype.Timestamptz `json:"hour"`
Iata string `json:"iata"`
ScopeID int32 `json:"scope_id"`
ObserverID uuid.UUID `json:"observer_id"`
}
type AnalyticsHourlyScopeSet struct {
Hour pgtype.Timestamptz `json:"hour"`
Iatas []string `json:"iatas"`
+5 -1
View File
@@ -24,7 +24,9 @@ WITH readings AS (
SELECT (grouping(hour) = 0)::boolean AS is_hourly,
COALESCE(hour, 'epoch'::timestamptz)::timestamptz AS hour,
category, hash_bytes, COALESCE(entries, 0)::integer AS entries,
sum(receptions)::bigint AS receptions
sum(receptions)::bigint AS receptions,
-- Non-routed categories store entries = 0, so an hour of only empty paths gives 0.
COALESCE(max(entries) FILTER (WHERE receptions > 0), 0)::integer AS max_entries
FROM readings
GROUP BY GROUPING SETS ((category, hash_bytes, entries), (hour, category, hash_bytes))
ORDER BY is_hourly, hour, category, hash_bytes, entries
@@ -43,6 +45,7 @@ type GetPathStatsRow struct {
HashBytes int32 `json:"hash_bytes"`
Entries int32 `json:"entries"`
Receptions int64 `json:"receptions"`
MaxEntries int32 `json:"max_entries"`
}
// All HTTP aggregates read the hourly rollup, never observations.
@@ -62,6 +65,7 @@ func (q *Queries) GetPathStats(ctx context.Context, arg GetPathStatsParams) ([]G
&i.HashBytes,
&i.Entries,
&i.Receptions,
&i.MaxEntries,
); err != nil {
return nil, err
}
+5
View File
@@ -122,6 +122,9 @@ type Querier interface {
// Packets since the given hour come from the IATA-set rollup (counted once per hour heard).
// Observer and node counts are current memberships; observers filter by their latest IATA.
GetScopeStats(ctx context.Context, arg GetScopeStatsParams) ([]GetScopeStatsRow, error)
// GetScopeStats packet counts split by hour (same window and IATA-set filter), with the
// distinct observers and advertising nodes active in each scope that hour. Empty hours omitted.
GetScopeStatsHourly(ctx context.Context, arg GetScopeStatsHourlyParams) ([]GetScopeStatsHourlyRow, error)
// Read the hourly rollup, never observations on an HTTP request.
// Weight averages by sample counts instead of averaging regional/hourly means.
GetSignalStats(ctx context.Context, arg GetSignalStatsParams) ([]GetSignalStatsRow, error)
@@ -315,6 +318,8 @@ type Querier interface {
// category: 0 = routed, 1 = zero-hop, 2 = TRACE, 3 = invalid (meshcore-go IsValidPathLen).
RollPaths(ctx context.Context, hour pgtype.Timestamptz) error
RollPayloadBreakdown(ctx context.Context, hour pgtype.Timestamptz) error
RollScopeNodes(ctx context.Context, hour pgtype.Timestamptz) error
RollScopeObservers(ctx context.Context, hour pgtype.Timestamptz) error
RollScopeSets(ctx context.Context, hour pgtype.Timestamptz) error
// kind = grouping(snr_bin, rssi_bin): 3 = per IATA totals, 1 = per SNR bin, 2 = per RSSI bin.
// rssi=0 AND snr=0 means "not reported".
+76
View File
@@ -1725,6 +1725,82 @@ func (q *Queries) GetScopeStats(ctx context.Context, arg GetScopeStatsParams) ([
return items, nil
}
const getScopeStatsHourly = `-- name: GetScopeStatsHourly :many
WITH pk AS (
SELECT s.scope_id, s.hour, SUM(s.packets)::bigint AS n
FROM analytics_hourly_scope_sets s
WHERE s.hour >= $1::timestamptz
AND (COALESCE(cardinality($2::bpchar[]), 0) = 0 OR s.iatas && $2::bpchar[])
GROUP BY s.scope_id, s.hour
), ob AS (
SELECT o.scope_id, o.hour, COUNT(DISTINCT o.observer_id)::bigint AS n
FROM analytics_hourly_scope_observers o
WHERE o.hour >= $1::timestamptz
AND (COALESCE(cardinality($2::bpchar[]), 0) = 0 OR o.iata = ANY($2::bpchar[]))
GROUP BY o.scope_id, o.hour
), nd AS (
SELECT d.scope_id, d.hour, COUNT(DISTINCT d.origin_pubkey)::bigint AS n
FROM analytics_hourly_scope_nodes d
WHERE d.hour >= $1::timestamptz
AND (COALESCE(cardinality($2::bpchar[]), 0) = 0 OR d.iata = ANY($2::bpchar[]))
GROUP BY d.scope_id, d.hour
), k AS (
SELECT scope_id, hour FROM pk UNION SELECT scope_id, hour FROM ob UNION SELECT scope_id, hour FROM nd
)
SELECT ts.name, k.hour,
COALESCE(pk.n, 0)::bigint AS packets,
COALESCE(ob.n, 0)::bigint AS observers,
COALESCE(nd.n, 0)::bigint AS nodes
FROM k
JOIN transport_scopes ts ON ts.id = k.scope_id
LEFT JOIN pk ON pk.scope_id = k.scope_id AND pk.hour = k.hour
LEFT JOIN ob ON ob.scope_id = k.scope_id AND ob.hour = k.hour
LEFT JOIN nd ON nd.scope_id = k.scope_id AND nd.hour = k.hour
WHERE COALESCE(pk.n, 0) > 0 OR COALESCE(ob.n, 0) > 0 OR COALESCE(nd.n, 0) > 0
ORDER BY ts.name, k.hour
`
type GetScopeStatsHourlyParams struct {
Since pgtype.Timestamptz `json:"since"`
Iatas []string `json:"iatas"`
}
type GetScopeStatsHourlyRow struct {
Name string `json:"name"`
Hour pgtype.Timestamptz `json:"hour"`
Packets int64 `json:"packets"`
Observers int64 `json:"observers"`
Nodes int64 `json:"nodes"`
}
// GetScopeStats packet counts split by hour (same window and IATA-set filter), with the
// distinct observers and advertising nodes active in each scope that hour. Empty hours omitted.
func (q *Queries) GetScopeStatsHourly(ctx context.Context, arg GetScopeStatsHourlyParams) ([]GetScopeStatsHourlyRow, error) {
rows, err := q.db.Query(ctx, getScopeStatsHourly, arg.Since, arg.Iatas)
if err != nil {
return nil, err
}
defer rows.Close()
items := []GetScopeStatsHourlyRow{}
for rows.Next() {
var i GetScopeStatsHourlyRow
if err := rows.Scan(
&i.Name,
&i.Hour,
&i.Packets,
&i.Observers,
&i.Nodes,
); err != nil {
return nil, err
}
items = append(items, i)
}
if err := rows.Err(); err != nil {
return nil, err
}
return items, nil
}
const getStatsClockDrift = `-- name: GetStatsClockDrift :many
WITH page AS (
SELECT
+31 -3
View File
@@ -24,6 +24,8 @@ WITH a AS (DELETE FROM analytics_hourly_iata_observations ta WHERE ta.hour < $1:
j AS (DELETE FROM analytics_hourly_packet_sets tj WHERE tj.hour < $1::timestamptz),
k AS (DELETE FROM analytics_hourly_scope_sets tk WHERE tk.hour < $1::timestamptz),
l AS (DELETE FROM analytics_dirty_hours tl WHERE tl.hour < $1::timestamptz),
n AS (DELETE FROM analytics_hourly_scope_observers tn WHERE tn.hour < $1::timestamptz),
o AS (DELETE FROM analytics_hourly_scope_nodes tob WHERE tob.hour < $1::timestamptz),
m AS (DELETE FROM analytics_rollup_hours tm WHERE tm.hour < $1::timestamptz RETURNING 1)
UPDATE analytics_raw_state SET coverage = coverage + 1 WHERE EXISTS (SELECT 1 FROM m)
`
@@ -44,8 +46,10 @@ WITH a AS (DELETE FROM analytics_hourly_iata_observations ta WHERE ta.hour = $1:
g AS (DELETE FROM analytics_hourly_advert_hearings tg WHERE tg.hour = $1::timestamptz),
h AS (DELETE FROM analytics_hourly_advert_sets th WHERE th.hour = $1::timestamptz),
i AS (DELETE FROM analytics_hourly_talker_sets ti WHERE ti.hour = $1::timestamptz),
j AS (DELETE FROM analytics_hourly_packet_sets tj WHERE tj.hour = $1::timestamptz)
DELETE FROM analytics_hourly_scope_sets tk WHERE tk.hour = $1::timestamptz
j AS (DELETE FROM analytics_hourly_packet_sets tj WHERE tj.hour = $1::timestamptz),
k AS (DELETE FROM analytics_hourly_scope_sets tk WHERE tk.hour = $1::timestamptz),
l AS (DELETE FROM analytics_hourly_scope_observers tl WHERE tl.hour = $1::timestamptz)
DELETE FROM analytics_hourly_scope_nodes tm WHERE tm.hour = $1::timestamptz
`
func (q *Queries) DeleteRollupHour(ctx context.Context, hour pgtype.Timestamptz) error {
@@ -379,6 +383,28 @@ func (q *Queries) RollPayloadBreakdown(ctx context.Context, hour pgtype.Timestam
return err
}
const rollScopeNodes = `-- name: RollScopeNodes :exec
INSERT INTO analytics_hourly_scope_nodes (hour, iata, scope_id, origin_pubkey)
SELECT DISTINCT $1::timestamptz, iata, scope_id, origin_pubkey
FROM rollup_obs WHERE payload_type = 4 AND origin_pubkey IS NOT NULL AND scope_id IS NOT NULL
`
func (q *Queries) RollScopeNodes(ctx context.Context, hour pgtype.Timestamptz) error {
_, err := q.db.Exec(ctx, rollScopeNodes, hour)
return err
}
const rollScopeObservers = `-- name: RollScopeObservers :exec
INSERT INTO analytics_hourly_scope_observers (hour, iata, scope_id, observer_id)
SELECT DISTINCT $1::timestamptz, iata, scope_id, observer_id
FROM rollup_obs WHERE scope_id IS NOT NULL
`
func (q *Queries) RollScopeObservers(ctx context.Context, hour pgtype.Timestamptz) error {
_, err := q.db.Exec(ctx, rollScopeObservers, hour)
return err
}
const rollScopeSets = `-- name: RollScopeSets :exec
INSERT INTO analytics_hourly_scope_sets (hour, iatas, scope_id, packets)
SELECT $1::timestamptz, iatas, scope_id, count(*)
@@ -451,7 +477,9 @@ SELECT md5(concat_ws('|',
(SELECT string_agg(t::text, ',' ORDER BY t::text) FROM analytics_hourly_advert_sets t WHERE t.hour = $1::timestamptz),
(SELECT string_agg(t::text, ',' ORDER BY t::text) FROM analytics_hourly_talker_sets t WHERE t.hour = $1::timestamptz),
(SELECT string_agg(t::text, ',' ORDER BY t::text) FROM analytics_hourly_packet_sets t WHERE t.hour = $1::timestamptz),
(SELECT string_agg(t::text, ',' ORDER BY t::text) FROM analytics_hourly_scope_sets t WHERE t.hour = $1::timestamptz)
(SELECT string_agg(t::text, ',' ORDER BY t::text) FROM analytics_hourly_scope_sets t WHERE t.hour = $1::timestamptz),
(SELECT string_agg(t::text, ',' ORDER BY t::text) FROM analytics_hourly_scope_observers t WHERE t.hour = $1::timestamptz),
(SELECT string_agg(t::text, ',' ORDER BY t::text) FROM analytics_hourly_scope_nodes t WHERE t.hour = $1::timestamptz)
))::text AS content_hash
`
+15 -1
View File
@@ -225,17 +225,31 @@ func (s *Store) GetRadioPresets(ctx context.Context, preset string, iatas []stri
}
func (s *Store) GetScopeStats(ctx context.Context, iatas []string, since time.Time) ([]api.ScopeStats, error) {
rows, err := s.q.GetScopeStats(ctx, sqlc.GetScopeStatsParams{Since: rollupSince(since, 7*24*time.Hour), Iatas: iatas})
start := rollupSince(since, 7*24*time.Hour)
rows, err := s.q.GetScopeStats(ctx, sqlc.GetScopeStatsParams{Since: start, Iatas: iatas})
if err != nil {
return nil, err
}
hours, err := s.q.GetScopeStatsHourly(ctx, sqlc.GetScopeStatsHourlyParams{Since: start, Iatas: iatas})
if err != nil {
return nil, err
}
byScope := make(map[string][]api.ScopeHour)
for _, h := range hours {
byScope[h.Name] = append(byScope[h.Name], api.ScopeHour{Hour: h.Hour.Time.UnixMilli(), Packets: h.Packets, Observers: h.Observers, Nodes: h.Nodes})
}
items := make([]api.ScopeStats, 0, len(rows))
for _, r := range rows {
hourly := byScope[r.Name]
if hourly == nil {
hourly = []api.ScopeHour{}
}
items = append(items, api.ScopeStats{
Name: r.Name,
PacketCount: r.PacketCount,
ObserverCount: r.ObserverCount,
NodeCount: r.NodeCount,
Hourly: hourly,
})
}
return items, nil
+16 -2
View File
@@ -232,6 +232,14 @@ func TestGetScopeStats(t *testing.T) {
GetScopeStats(gomock.Any(), gomock.Any()).
Return([]sqlc.GetScopeStatsRow{
{Name: "default", PacketCount: 100, ObserverCount: 5, NodeCount: 20},
{Name: "quiet"},
}, nil)
hour := time.Date(2026, 1, 1, 3, 0, 0, 0, time.UTC)
mock.EXPECT().
GetScopeStatsHourly(gomock.Any(), gomock.Any()).
Return([]sqlc.GetScopeStatsHourlyRow{
{Name: "default", Hour: pgtype.Timestamptz{Time: hour, Valid: true}, Packets: 60},
{Name: "default", Hour: pgtype.Timestamptz{Time: hour.Add(time.Hour), Valid: true}, Packets: 40},
}, nil)
store := &Store{q: mock}
@@ -239,8 +247,14 @@ func TestGetScopeStats(t *testing.T) {
if err != nil {
t.Fatalf("unexpected error: %v", err)
}
if len(items) != 1 {
t.Fatalf("expected 1 item, got %d", len(items))
if len(items) != 2 {
t.Fatalf("expected 2 items, got %d", len(items))
}
if h := items[0].Hourly; len(h) != 2 || h[0].Hour != hour.UnixMilli() || h[0].Packets+h[1].Packets != items[0].PacketCount {
t.Errorf("hourly %+v doesn't split packetCount %d", h, items[0].PacketCount)
}
if items[1].Hourly == nil || len(items[1].Hourly) != 0 {
t.Errorf("a scope without packets must have hourly [], got %#v", items[1].Hourly)
}
if items[0].Name != "default" {
t.Errorf("expected Name default, got %s", items[0].Name)