diff --git a/db/migrations/001_baseline.sql b/db/migrations/001_baseline.sql index 00af459..7dd67ed 100644 --- a/db/migrations/001_baseline.sql +++ b/db/migrations/001_baseline.sql @@ -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')), diff --git a/db/paths.go b/db/paths.go index 19cffd8..2189f92 100644 --- a/db/paths.go +++ b/db/paths.go @@ -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 { diff --git a/db/paths_integration_test.go b/db/paths_integration_test.go index 656c48f..fbec87f 100644 --- a/db/paths_integration_test.go +++ b/db/paths_integration_test.go @@ -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 diff --git a/db/paths_test.go b/db/paths_test.go index beb2f58..3b13606 100644 --- a/db/paths_test.go +++ b/db/paths_test.go @@ -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) + } +} diff --git a/db/queries/paths.sql b/db/queries/paths.sql index 8cdaf51..effe72f 100644 --- a/db/queries/paths.sql +++ b/db/queries/paths.sql @@ -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; diff --git a/db/queries/queries.sql b/db/queries/queries.sql index 349d621..dacbb98 100644 --- a/db/queries/queries.sql +++ b/db/queries/queries.sql @@ -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. diff --git a/db/queries/rollup.sql b/db/queries/rollup.sql index 5b33676..889199c 100644 --- a/db/queries/rollup.sql +++ b/db/queries/rollup.sql @@ -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); diff --git a/db/rollup.go b/db/rollup.go index 3c2cab3..fc8e3fd 100644 --- a/db/rollup.go +++ b/db/rollup.go @@ -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) diff --git a/db/rollup_integration_test.go b/db/rollup_integration_test.go index a5cace6..d71d73b 100644 --- a/db/rollup_integration_test.go +++ b/db/rollup_integration_test.go @@ -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 diff --git a/db/scope_stats_integration_test.go b/db/scope_stats_integration_test.go index 207e09d..8a07435 100644 --- a/db/scope_stats_integration_test.go +++ b/db/scope_stats_integration_test.go @@ -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) diff --git a/db/sqlc/mock/querier.go b/db/sqlc/mock/querier.go index 2b436d9..e61bc78 100644 --- a/db/sqlc/mock/querier.go +++ b/db/sqlc/mock/querier.go @@ -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() diff --git a/db/sqlc/models.go b/db/sqlc/models.go index 709db8f..78e1430 100644 --- a/db/sqlc/models.go +++ b/db/sqlc/models.go @@ -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"` diff --git a/db/sqlc/paths.sql.go b/db/sqlc/paths.sql.go index f5c24f1..5c2f342 100644 --- a/db/sqlc/paths.sql.go +++ b/db/sqlc/paths.sql.go @@ -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 } diff --git a/db/sqlc/querier.go b/db/sqlc/querier.go index 1eaf656..834628f 100644 --- a/db/sqlc/querier.go +++ b/db/sqlc/querier.go @@ -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". diff --git a/db/sqlc/queries.sql.go b/db/sqlc/queries.sql.go index 0a3b572..7c6d7dc 100644 --- a/db/sqlc/queries.sql.go +++ b/db/sqlc/queries.sql.go @@ -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 diff --git a/db/sqlc/rollup.sql.go b/db/sqlc/rollup.sql.go index db8ee8f..02f7fd0 100644 --- a/db/sqlc/rollup.sql.go +++ b/db/sqlc/rollup.sql.go @@ -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 ` diff --git a/db/stats.go b/db/stats.go index 5bda12c..8d614ca 100644 --- a/db/stats.go +++ b/db/stats.go @@ -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 diff --git a/db/stats_test.go b/db/stats_test.go index 88e978b..d70a1f3 100644 --- a/db/stats_test.go +++ b/db/stats_test.go @@ -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) diff --git a/docs/docs.go b/docs/docs.go index 22831d4..c8b92de 100644 --- a/docs/docs.go +++ b/docs/docs.go @@ -2835,7 +2835,7 @@ const docTemplate = `{ }, "/stats/scopes": { "get": { - "description": "Packets are those heard since the window start, each counted once per hour however many of the requested IATAs heard it. Observers and nodes are current memberships: observers filter by the IATA they last reported from, nodes by their IATA memberships. With filters, lists only manual scopes configured for a matching region and imported scopes whose current MeshMapper catalogue includes a matching IATA; those remain listed with zero counts. An empty region returns an empty array.", + "description": "Packets are those heard since the window start, each counted once per hour however many of the requested IATAs heard it. Observers and nodes are current memberships: observers filter by the IATA they last reported from, nodes by their IATA memberships. hourly splits packetCount by UTC hour and adds the distinct observers and advertising nodes active in the scope each hour (empty hours omitted; outage gaps come from /stats/series hour status). With filters, lists only manual scopes configured for a matching region and imported scopes whose current MeshMapper catalogue includes a matching IATA; those remain listed with zero counts. An empty region returns an empty array.", "produces": [ "application/json" ], @@ -4872,6 +4872,10 @@ const docTemplate = `{ "hour": { "type": "integer" }, + "maxEntries": { + "description": "longest path received that hour across the requested IATAs; 0 if only empty paths", + "type": "integer" + }, "oneByte": { "type": "integer" }, @@ -5205,9 +5209,36 @@ const docTemplate = `{ } } }, + "github_com_MeshCore-Beacon_beacon-server_internal_api.ScopeHour": { + "type": "object", + "properties": { + "hour": { + "description": "epoch ms, start of the UTC hour", + "type": "integer" + }, + "nodes": { + "description": "distinct nodes whose adverts were heard in the scope that hour", + "type": "integer" + }, + "observers": { + "description": "distinct observers that heard the scope's packets that hour", + "type": "integer" + }, + "packets": { + "type": "integer" + } + } + }, "github_com_MeshCore-Beacon_beacon-server_internal_api.ScopeStats": { "type": "object", "properties": { + "hourly": { + "description": "packetCount by hour, oldest first; hours with none omitted", + "type": "array", + "items": { + "$ref": "#/definitions/github_com_MeshCore-Beacon_beacon-server_internal_api.ScopeHour" + } + }, "name": { "description": "normalized scope name e.g. \"#bc\"", "type": "string" diff --git a/docs/historical-stats.md b/docs/historical-stats.md index 72ad3ac..8c8ebff 100644 --- a/docs/historical-stats.md +++ b/docs/historical-stats.md @@ -48,6 +48,7 @@ sub-hour observer activity. | Observations, receptions, payload counts | Sum | | Unique packets, adverts, messages, scoped packets | Sum of hourly IATA-set counts: each is counted once per hour it was heard, so one heard across an hour boundary counts twice (measured +1.1%) | | Active observers, IATAs, scopes | Distinct across the window | +| Active scope observers and nodes (per scope, per hour) | Distinct within the hour, across the requested IATAs | | Max path entries | Max | | SNR / RSSI averages | Σ sum / Σ samples, never an average of averages | diff --git a/docs/swagger.json b/docs/swagger.json index 52c2c9d..7675cf5 100644 --- a/docs/swagger.json +++ b/docs/swagger.json @@ -2833,7 +2833,7 @@ }, "/stats/scopes": { "get": { - "description": "Packets are those heard since the window start, each counted once per hour however many of the requested IATAs heard it. Observers and nodes are current memberships: observers filter by the IATA they last reported from, nodes by their IATA memberships. With filters, lists only manual scopes configured for a matching region and imported scopes whose current MeshMapper catalogue includes a matching IATA; those remain listed with zero counts. An empty region returns an empty array.", + "description": "Packets are those heard since the window start, each counted once per hour however many of the requested IATAs heard it. Observers and nodes are current memberships: observers filter by the IATA they last reported from, nodes by their IATA memberships. hourly splits packetCount by UTC hour and adds the distinct observers and advertising nodes active in the scope each hour (empty hours omitted; outage gaps come from /stats/series hour status). With filters, lists only manual scopes configured for a matching region and imported scopes whose current MeshMapper catalogue includes a matching IATA; those remain listed with zero counts. An empty region returns an empty array.", "produces": [ "application/json" ], @@ -4870,6 +4870,10 @@ "hour": { "type": "integer" }, + "maxEntries": { + "description": "longest path received that hour across the requested IATAs; 0 if only empty paths", + "type": "integer" + }, "oneByte": { "type": "integer" }, @@ -5203,9 +5207,36 @@ } } }, + "github_com_MeshCore-Beacon_beacon-server_internal_api.ScopeHour": { + "type": "object", + "properties": { + "hour": { + "description": "epoch ms, start of the UTC hour", + "type": "integer" + }, + "nodes": { + "description": "distinct nodes whose adverts were heard in the scope that hour", + "type": "integer" + }, + "observers": { + "description": "distinct observers that heard the scope's packets that hour", + "type": "integer" + }, + "packets": { + "type": "integer" + } + } + }, "github_com_MeshCore-Beacon_beacon-server_internal_api.ScopeStats": { "type": "object", "properties": { + "hourly": { + "description": "packetCount by hour, oldest first; hours with none omitted", + "type": "array", + "items": { + "$ref": "#/definitions/github_com_MeshCore-Beacon_beacon-server_internal_api.ScopeHour" + } + }, "name": { "description": "normalized scope name e.g. \"#bc\"", "type": "string" diff --git a/docs/swagger.yaml b/docs/swagger.yaml index 95e7171..5ec66b4 100644 --- a/docs/swagger.yaml +++ b/docs/swagger.yaml @@ -1021,6 +1021,10 @@ definitions: type: integer hour: type: integer + maxEntries: + description: longest path received that hour across the requested IATAs; 0 + if only empty paths + type: integer oneByte: type: integer receptions: @@ -1242,8 +1246,27 @@ definitions: packetCount: type: integer type: object + github_com_MeshCore-Beacon_beacon-server_internal_api.ScopeHour: + properties: + hour: + description: epoch ms, start of the UTC hour + type: integer + nodes: + description: distinct nodes whose adverts were heard in the scope that hour + type: integer + observers: + description: distinct observers that heard the scope's packets that hour + type: integer + packets: + type: integer + type: object github_com_MeshCore-Beacon_beacon-server_internal_api.ScopeStats: properties: + hourly: + description: packetCount by hour, oldest first; hours with none omitted + items: + $ref: '#/definitions/github_com_MeshCore-Beacon_beacon-server_internal_api.ScopeHour' + type: array name: description: normalized scope name e.g. "#bc" type: string @@ -3576,10 +3599,12 @@ paths: description: 'Packets are those heard since the window start, each counted once per hour however many of the requested IATAs heard it. Observers and nodes are current memberships: observers filter by the IATA they last reported from, - nodes by their IATA memberships. With filters, lists only manual scopes configured - for a matching region and imported scopes whose current MeshMapper catalogue - includes a matching IATA; those remain listed with zero counts. An empty region - returns an empty array.' + nodes by their IATA memberships. hourly splits packetCount by UTC hour and + adds the distinct observers and advertising nodes active in the scope each + hour (empty hours omitted; outage gaps come from /stats/series hour status). + With filters, lists only manual scopes configured for a matching region and + imported scopes whose current MeshMapper catalogue includes a matching IATA; + those remain listed with zero counts. An empty region returns an empty array.' parameters: - description: Comma-separated IATA codes in: query diff --git a/internal/api/handlers/stats.go b/internal/api/handlers/stats.go index dc6216f..43164d0 100644 --- a/internal/api/handlers/stats.go +++ b/internal/api/handlers/stats.go @@ -457,7 +457,7 @@ func getStatsRadioPresets(reader api.Reader) http.HandlerFunc { // getStatsScopes godoc // // @Summary Scope statistics -// @Description Packets are those heard since the window start, each counted once per hour however many of the requested IATAs heard it. Observers and nodes are current memberships: observers filter by the IATA they last reported from, nodes by their IATA memberships. With filters, lists only manual scopes configured for a matching region and imported scopes whose current MeshMapper catalogue includes a matching IATA; those remain listed with zero counts. An empty region returns an empty array. +// @Description Packets are those heard since the window start, each counted once per hour however many of the requested IATAs heard it. Observers and nodes are current memberships: observers filter by the IATA they last reported from, nodes by their IATA memberships. hourly splits packetCount by UTC hour and adds the distinct observers and advertising nodes active in the scope each hour (empty hours omitted; outage gaps come from /stats/series hour status). With filters, lists only manual scopes configured for a matching region and imported scopes whose current MeshMapper catalogue includes a matching IATA; those remain listed with zero counts. An empty region returns an empty array. // @Tags Stats // @Produce json // @Param iatas query string false "Comma-separated IATA codes" diff --git a/internal/api/paths.go b/internal/api/paths.go index 86c3a00..dad76d1 100644 --- a/internal/api/paths.go +++ b/internal/api/paths.go @@ -43,4 +43,5 @@ type PathHour struct { Empty int64 `json:"empty"` Trace int64 `json:"trace"` Unclassified int64 `json:"unclassified"` + MaxEntries int32 `json:"maxEntries"` // longest path received that hour across the requested IATAs; 0 if only empty paths } diff --git a/internal/api/stats.go b/internal/api/stats.go index bb1e8eb..fe3b697 100644 --- a/internal/api/stats.go +++ b/internal/api/stats.go @@ -92,10 +92,19 @@ type PayloadBreakdownItem struct { // ScopeStats represents aggregate statistics for a single transport scope. type ScopeStats struct { - Name string `json:"name"` // normalized scope name e.g. "#bc" - PacketCount int64 `json:"packetCount"` // distinct packets matched to this scope - ObserverCount int64 `json:"observerCount"` // distinct observers that forwarded packets in this scope - NodeCount int64 `json:"nodeCount"` // distinct nodes with this as their default scope + Name string `json:"name"` // normalized scope name e.g. "#bc" + PacketCount int64 `json:"packetCount"` // distinct packets matched to this scope + ObserverCount int64 `json:"observerCount"` // distinct observers that forwarded packets in this scope + NodeCount int64 `json:"nodeCount"` // distinct nodes with this as their default scope + Hourly []ScopeHour `json:"hourly"` // packetCount by hour, oldest first; hours with none omitted +} + +// ScopeHour is one UTC hour of a scope's activity; a scope's hours' packets sum to its packetCount. +type ScopeHour struct { + Hour int64 `json:"hour"` // epoch ms, start of the UTC hour + Packets int64 `json:"packets"` + Observers int64 `json:"observers"` // distinct observers that heard the scope's packets that hour + Nodes int64 `json:"nodes"` // distinct nodes whose adverts were heard in the scope that hour } // TopNode is a node ranked by how often its adverts were heard in the window. diff --git a/internal/cache/rollup_stats.go b/internal/cache/rollup_stats.go index 6272fb1..6ab0148 100644 --- a/internal/cache/rollup_stats.go +++ b/internal/cache/rollup_stats.go @@ -158,7 +158,8 @@ func (cr *CachedReader) GetStatsTopTalkers(ctx context.Context, iatas []string, // GetScopeStats implements [api.Reader]. func (cr *CachedReader) GetScopeStats(ctx context.Context, iatas []string, since time.Time) ([]api.ScopeStats, error) { - return cachedRollup(ctx, cr, keyScopeStats+":", iatas, func(ctx context.Context) ([]api.ScopeStats, error) { + // The suffix changes with the hourly shape so older cached entries aren't served. + return cachedRollup(ctx, cr, keyScopeStats+":h2:", iatas, func(ctx context.Context) ([]api.ScopeStats, error) { return cr.inner.GetScopeStats(ctx, iatas, since) }, hourKey(since)) }