From d8ff7d0542c9bb28242e0a58e2bc43f853a44238 Mon Sep 17 00:00:00 2001 From: MrAlders0n Date: Thu, 23 Jul 2026 22:13:29 -0400 Subject: [PATCH] track payload_type on observations and serve the breakdown from a matview --- .../015_observations_payload_type.sql | 17 +++++++ db/migrations/016_mv_payload_breakdown.sql | 15 ++++++ db/packets.go | 1 + db/queries/queries.sql | 22 ++++---- db/sqlc/mock/querier.go | 22 ++++++-- db/sqlc/models.go | 7 +++ db/sqlc/querier.go | 5 +- db/sqlc/queries.sql.go | 50 +++++++++++-------- db/stats.go | 22 ++++---- internal/background/tasks.go | 3 ++ internal/ingest/packet.go | 2 + 11 files changed, 119 insertions(+), 47 deletions(-) create mode 100644 db/migrations/015_observations_payload_type.sql create mode 100644 db/migrations/016_mv_payload_breakdown.sql diff --git a/db/migrations/015_observations_payload_type.sql b/db/migrations/015_observations_payload_type.sql new file mode 100644 index 0000000..e3bb79b --- /dev/null +++ b/db/migrations/015_observations_payload_type.sql @@ -0,0 +1,17 @@ +-- Copy payload_type onto the observation (like source_broker) so the breakdown +-- drops its join back to packets, which was overrunning /dev/shm and 500ing. + +ALTER TABLE packet_observations ADD COLUMN payload_type SMALLINT; + +-- Backfill the last 7 days only (all the breakdown reads); parallelism off so +-- the join spills to disk, not /dev/shm. +SET max_parallel_workers_per_gather = 0; + +UPDATE packet_observations po +SET payload_type = p.payload_type +FROM packets p +WHERE p.packet_hash = po.packet_hash + AND po.heard_at > NOW() - INTERVAL '7 days' + AND po.payload_type IS NULL; + +RESET max_parallel_workers_per_gather; diff --git a/db/migrations/016_mv_payload_breakdown.sql b/db/migrations/016_mv_payload_breakdown.sql new file mode 100644 index 0000000..0710015 --- /dev/null +++ b/db/migrations/016_mv_payload_breakdown.sql @@ -0,0 +1,15 @@ +-- Precomputed payload breakdown so the request reads a small table instead of +-- scanning a week of observations. + +CREATE MATERIALIZED VIEW mv_payload_breakdown_by_iata AS +SELECT + iata, + payload_type, + COUNT(*) AS count +FROM packet_observations +WHERE heard_at > NOW() - INTERVAL '7 days' + AND payload_type IS NOT NULL +GROUP BY iata, payload_type; + +CREATE UNIQUE INDEX idx_mv_payload_breakdown + ON mv_payload_breakdown_by_iata(iata, payload_type); diff --git a/db/packets.go b/db/packets.go index 66a0be6..68bd34c 100644 --- a/db/packets.go +++ b/db/packets.go @@ -505,6 +505,7 @@ func (s *Store) InsertObservation(ctx context.Context, o ingest.InsertObservatio BandwidthKhz: &o.BandwidthKHz, CodingRate: &o.CodingRate, SourceBroker: &o.SourceBroker, + PayloadType: &o.PayloadType, } row, err := s.q.InsertObservation(ctx, params) if errors.Is(err, pgx.ErrNoRows) { diff --git a/db/queries/queries.sql b/db/queries/queries.sql index bb8008b..b203e4c 100644 --- a/db/queries/queries.sql +++ b/db/queries/queries.sql @@ -523,9 +523,10 @@ INSERT INTO packet_observations ( spread_factor, bandwidth_khz, coding_rate, - source_broker + source_broker, + payload_type ) VALUES ( - $1, $2, $3, $4, $5, $6, $7, $8, $9, $10, $11, $12, $13, $14, $15, $16 + $1, $2, $3, $4, $5, $6, $7, $8, $9, $10, $11, $12, $13, $14, $15, $16, $17 ) ON CONFLICT (packet_hash, observer_id) DO NOTHING RETURNING *; @@ -821,15 +822,13 @@ ORDER BY observation_count DESC LIMIT $2; -- name: GetStatsPayloadBreakdown :many --- Returns observation counts grouped by payload type for the given window and IATA. +-- Payload-type counts for the IATA, from the precomputed view. SELECT - p.payload_type, - COUNT(*) AS count -FROM packet_observations po -JOIN packets p ON p.packet_hash = po.packet_hash -WHERE po.heard_at > $1 - AND (COALESCE(cardinality($2::bpchar[]), 0) = 0 OR po.iata = ANY($2::bpchar[])) -GROUP BY p.payload_type + payload_type, + SUM(count)::bigint AS count +FROM mv_payload_breakdown_by_iata +WHERE (COALESCE(cardinality($1::bpchar[]), 0) = 0 OR iata = ANY($1::bpchar[])) +GROUP BY payload_type ORDER BY count DESC; -- name: GetStatsNodeTypes :many @@ -1141,6 +1140,9 @@ REFRESH MATERIALIZED VIEW CONCURRENTLY mv_hourly_iata_stats; -- name: RefreshTopNodes :exec REFRESH MATERIALIZED VIEW CONCURRENTLY mv_top_nodes_by_iata; +-- name: RefreshPayloadBreakdown :exec +REFRESH MATERIALIZED VIEW CONCURRENTLY mv_payload_breakdown_by_iata; + -- name: RefreshRadioPresets :exec REFRESH MATERIALIZED VIEW CONCURRENTLY mv_radio_presets; diff --git a/db/sqlc/mock/querier.go b/db/sqlc/mock/querier.go index 22f0983..aa8d1da 100644 --- a/db/sqlc/mock/querier.go +++ b/db/sqlc/mock/querier.go @@ -550,18 +550,18 @@ func (mr *MockQuerierMockRecorder) GetStatsOverview(ctx, dollar_1 any) *gomock.C } // GetStatsPayloadBreakdown mocks base method. -func (m *MockQuerier) GetStatsPayloadBreakdown(ctx context.Context, arg db.GetStatsPayloadBreakdownParams) ([]db.GetStatsPayloadBreakdownRow, error) { +func (m *MockQuerier) GetStatsPayloadBreakdown(ctx context.Context, dollar_1 []string) ([]db.GetStatsPayloadBreakdownRow, error) { m.ctrl.T.Helper() - ret := m.ctrl.Call(m, "GetStatsPayloadBreakdown", ctx, arg) + ret := m.ctrl.Call(m, "GetStatsPayloadBreakdown", ctx, dollar_1) ret0, _ := ret[0].([]db.GetStatsPayloadBreakdownRow) ret1, _ := ret[1].(error) return ret0, ret1 } // GetStatsPayloadBreakdown indicates an expected call of GetStatsPayloadBreakdown. -func (mr *MockQuerierMockRecorder) GetStatsPayloadBreakdown(ctx, arg any) *gomock.Call { +func (mr *MockQuerierMockRecorder) GetStatsPayloadBreakdown(ctx, dollar_1 any) *gomock.Call { mr.mock.ctrl.T.Helper() - return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "GetStatsPayloadBreakdown", reflect.TypeOf((*MockQuerier)(nil).GetStatsPayloadBreakdown), ctx, arg) + return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "GetStatsPayloadBreakdown", reflect.TypeOf((*MockQuerier)(nil).GetStatsPayloadBreakdown), ctx, dollar_1) } // GetStatsTopAdvertisers mocks base method. @@ -995,6 +995,20 @@ func (mr *MockQuerierMockRecorder) RefreshHourlyStats(ctx any) *gomock.Call { return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "RefreshHourlyStats", reflect.TypeOf((*MockQuerier)(nil).RefreshHourlyStats), ctx) } +// RefreshPayloadBreakdown mocks base method. +func (m *MockQuerier) RefreshPayloadBreakdown(ctx context.Context) error { + m.ctrl.T.Helper() + ret := m.ctrl.Call(m, "RefreshPayloadBreakdown", ctx) + ret0, _ := ret[0].(error) + return ret0 +} + +// RefreshPayloadBreakdown indicates an expected call of RefreshPayloadBreakdown. +func (mr *MockQuerierMockRecorder) RefreshPayloadBreakdown(ctx any) *gomock.Call { + mr.mock.ctrl.T.Helper() + return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "RefreshPayloadBreakdown", reflect.TypeOf((*MockQuerier)(nil).RefreshPayloadBreakdown), ctx) +} + // RefreshRadioPresets mocks base method. func (m *MockQuerier) RefreshRadioPresets(ctx context.Context) error { m.ctrl.T.Helper() diff --git a/db/sqlc/models.go b/db/sqlc/models.go index 6e88f25..388fc85 100644 --- a/db/sqlc/models.go +++ b/db/sqlc/models.go @@ -74,6 +74,12 @@ type MvHourlyIataStat struct { ActiveObservers int64 `json:"active_observers"` } +type MvPayloadBreakdownByIatum struct { + Iata string `json:"iata"` + PayloadType *int16 `json:"payload_type"` + Count int64 `json:"count"` +} + type MvRadioPreset struct { Preset string `json:"preset"` Iata string `json:"iata"` @@ -250,6 +256,7 @@ type PacketObservation struct { BandwidthKhz *float32 `json:"bandwidth_khz"` CodingRate *int16 `json:"coding_rate"` SourceBroker *string `json:"source_broker"` + PayloadType *int16 `json:"payload_type"` } type Region struct { diff --git a/db/sqlc/querier.go b/db/sqlc/querier.go index efb1ba2..bb1712a 100644 --- a/db/sqlc/querier.go +++ b/db/sqlc/querier.go @@ -60,8 +60,8 @@ type Querier interface { // STATS // ============================================================ GetStatsOverview(ctx context.Context, dollar_1 []string) (GetStatsOverviewRow, error) - // Returns observation counts grouped by payload type for the given window and IATA. - GetStatsPayloadBreakdown(ctx context.Context, arg GetStatsPayloadBreakdownParams) ([]GetStatsPayloadBreakdownRow, error) + // Payload-type counts for the IATA, from the precomputed view. + GetStatsPayloadBreakdown(ctx context.Context, dollar_1 []string) ([]GetStatsPayloadBreakdownRow, error) // Returns the top N nodes by distinct ADVERT packet count for the given window and IATA. // COUNT(DISTINCT p.packet_hash) rather than COUNT(*): the same advert broadcast is commonly // heard by more than one observer, and each hearing is its own packet_observations row -- @@ -156,6 +156,7 @@ type Querier interface { // that IATA, or where any hop's prefix_4 is now ambiguous (matches >1 node). ReconfirmRoutes(ctx context.Context) error RefreshHourlyStats(ctx context.Context) error + RefreshPayloadBreakdown(ctx context.Context) error RefreshRadioPresets(ctx context.Context) error RefreshTopNodes(ctx context.Context) error // ============================================================ diff --git a/db/sqlc/queries.sql.go b/db/sqlc/queries.sql.go index 787f5cd..6871173 100644 --- a/db/sqlc/queries.sql.go +++ b/db/sqlc/queries.sql.go @@ -1197,29 +1197,22 @@ func (q *Queries) GetStatsOverview(ctx context.Context, dollar_1 []string) (GetS const getStatsPayloadBreakdown = `-- name: GetStatsPayloadBreakdown :many SELECT - p.payload_type, - COUNT(*) AS count -FROM packet_observations po -JOIN packets p ON p.packet_hash = po.packet_hash -WHERE po.heard_at > $1 - AND (COALESCE(cardinality($2::bpchar[]), 0) = 0 OR po.iata = ANY($2::bpchar[])) -GROUP BY p.payload_type + payload_type, + SUM(count)::bigint AS count +FROM mv_payload_breakdown_by_iata +WHERE (COALESCE(cardinality($1::bpchar[]), 0) = 0 OR iata = ANY($1::bpchar[])) +GROUP BY payload_type ORDER BY count DESC ` -type GetStatsPayloadBreakdownParams struct { - HeardAt pgtype.Timestamptz `json:"heard_at"` - Column2 []string `json:"column_2"` -} - type GetStatsPayloadBreakdownRow struct { - PayloadType int16 `json:"payload_type"` - Count int64 `json:"count"` + PayloadType *int16 `json:"payload_type"` + Count int64 `json:"count"` } -// Returns observation counts grouped by payload type for the given window and IATA. -func (q *Queries) GetStatsPayloadBreakdown(ctx context.Context, arg GetStatsPayloadBreakdownParams) ([]GetStatsPayloadBreakdownRow, error) { - rows, err := q.db.Query(ctx, getStatsPayloadBreakdown, arg.HeardAt, arg.Column2) +// Payload-type counts for the IATA, from the precomputed view. +func (q *Queries) GetStatsPayloadBreakdown(ctx context.Context, dollar_1 []string) ([]GetStatsPayloadBreakdownRow, error) { + rows, err := q.db.Query(ctx, getStatsPayloadBreakdown, dollar_1) if err != nil { return nil, err } @@ -1551,12 +1544,13 @@ INSERT INTO packet_observations ( spread_factor, bandwidth_khz, coding_rate, - source_broker + source_broker, + payload_type ) VALUES ( - $1, $2, $3, $4, $5, $6, $7, $8, $9, $10, $11, $12, $13, $14, $15, $16 + $1, $2, $3, $4, $5, $6, $7, $8, $9, $10, $11, $12, $13, $14, $15, $16, $17 ) ON CONFLICT (packet_hash, observer_id) DO NOTHING -RETURNING id, packet_hash, observer_id, iata, heard_at, path_length_byte, hash_size, hop_count, path_bytes, rssi, snr, propagation_time_ms, radio_freq_mhz, spread_factor, bandwidth_khz, coding_rate, source_broker +RETURNING id, packet_hash, observer_id, iata, heard_at, path_length_byte, hash_size, hop_count, path_bytes, rssi, snr, propagation_time_ms, radio_freq_mhz, spread_factor, bandwidth_khz, coding_rate, source_broker, payload_type ` type InsertObservationParams struct { @@ -1576,6 +1570,7 @@ type InsertObservationParams struct { BandwidthKhz *float32 `json:"bandwidth_khz"` CodingRate *int16 `json:"coding_rate"` SourceBroker *string `json:"source_broker"` + PayloadType *int16 `json:"payload_type"` } // ============================================================ @@ -1599,6 +1594,7 @@ func (q *Queries) InsertObservation(ctx context.Context, arg InsertObservationPa arg.BandwidthKhz, arg.CodingRate, arg.SourceBroker, + arg.PayloadType, ) var i PacketObservation err := row.Scan( @@ -1619,6 +1615,7 @@ func (q *Queries) InsertObservation(ctx context.Context, arg InsertObservationPa &i.BandwidthKhz, &i.CodingRate, &i.SourceBroker, + &i.PayloadType, ) return i, err } @@ -2293,7 +2290,7 @@ func (q *Queries) ListNodes(ctx context.Context, arg ListNodesParams) ([]ListNod } const listObservationsForPacket = `-- name: ListObservationsForPacket :many -SELECT po.id, po.packet_hash, po.observer_id, po.iata, po.heard_at, po.path_length_byte, po.hash_size, po.hop_count, po.path_bytes, po.rssi, po.snr, po.propagation_time_ms, po.radio_freq_mhz, po.spread_factor, po.bandwidth_khz, po.coding_rate, po.source_broker, o.display_name AS observer_name +SELECT po.id, po.packet_hash, po.observer_id, po.iata, po.heard_at, po.path_length_byte, po.hash_size, po.hop_count, po.path_bytes, po.rssi, po.snr, po.propagation_time_ms, po.radio_freq_mhz, po.spread_factor, po.bandwidth_khz, po.coding_rate, po.source_broker, po.payload_type, o.display_name AS observer_name FROM packet_observations po LEFT JOIN observers o ON o.id = po.observer_id WHERE po.packet_hash = $1 @@ -2318,6 +2315,7 @@ type ListObservationsForPacketRow struct { BandwidthKhz *float32 `json:"bandwidth_khz"` CodingRate *int16 `json:"coding_rate"` SourceBroker *string `json:"source_broker"` + PayloadType *int16 `json:"payload_type"` ObserverName *string `json:"observer_name"` } @@ -2348,6 +2346,7 @@ func (q *Queries) ListObservationsForPacket(ctx context.Context, packetHash []by &i.BandwidthKhz, &i.CodingRate, &i.SourceBroker, + &i.PayloadType, &i.ObserverName, ); err != nil { return nil, err @@ -3073,6 +3072,15 @@ func (q *Queries) RefreshHourlyStats(ctx context.Context) error { return err } +const refreshPayloadBreakdown = `-- name: RefreshPayloadBreakdown :exec +REFRESH MATERIALIZED VIEW CONCURRENTLY mv_payload_breakdown_by_iata +` + +func (q *Queries) RefreshPayloadBreakdown(ctx context.Context) error { + _, err := q.db.Exec(ctx, refreshPayloadBreakdown) + return err +} + const refreshRadioPresets = `-- name: RefreshRadioPresets :exec REFRESH MATERIALIZED VIEW CONCURRENTLY mv_radio_presets ` diff --git a/db/stats.go b/db/stats.go index 930601b..a152ad6 100644 --- a/db/stats.go +++ b/db/stats.go @@ -51,22 +51,20 @@ func (s *Store) GetStatsObservations(ctx context.Context, iatas []string, since return points, nil } -func (s *Store) GetStatsPayloadBreakdown(ctx context.Context, iatas []string, since time.Time) ([]api.PayloadBreakdownItem, error) { - if since.IsZero() { - since = time.Now().Add(-24 * time.Hour) - } - rows, err := s.q.GetStatsPayloadBreakdown(ctx, sqlc.GetStatsPayloadBreakdownParams{ - HeardAt: pgtype.Timestamptz{Time: since, Valid: true}, - Column2: iatas, - }) +// since is unused: the view is a rolling 7-day window refreshed on a timer. +func (s *Store) GetStatsPayloadBreakdown(ctx context.Context, iatas []string, _ time.Time) ([]api.PayloadBreakdownItem, error) { + rows, err := s.q.GetStatsPayloadBreakdown(ctx, iatas) if err != nil { return nil, err } items := make([]api.PayloadBreakdownItem, 0, len(rows)) for _, v := range rows { + if v.PayloadType == nil { + continue + } items = append(items, api.PayloadBreakdownItem{ - PayloadType: v.PayloadType, - PayloadTypeName: api.PayloadTypeName(v.PayloadType), + PayloadType: *v.PayloadType, + PayloadTypeName: api.PayloadTypeName(*v.PayloadType), Count: v.Count, }) } @@ -244,6 +242,10 @@ func (s *Store) RefreshTopNodes(ctx context.Context) error { return s.q.RefreshTopNodes(ctx) } +func (s *Store) RefreshPayloadBreakdown(ctx context.Context) error { + return s.q.RefreshPayloadBreakdown(ctx) +} + func (s *Store) RefreshRadioPresets(ctx context.Context) error { return s.q.RefreshRadioPresets(ctx) } diff --git a/internal/background/tasks.go b/internal/background/tasks.go index 8824304..1966c6f 100644 --- a/internal/background/tasks.go +++ b/internal/background/tasks.go @@ -24,6 +24,9 @@ func ViewRefreshTask(store *db.Store, interval time.Duration) Task { if err := store.RefreshTopNodes(ctx); err != nil { log.Printf("background[view_refresh]: top nodes: %v", err) } + if err := store.RefreshPayloadBreakdown(ctx); err != nil { + log.Printf("background[view_refresh]: payload breakdown: %v", err) + } if err := store.RefreshRadioPresets(ctx); err != nil { log.Printf("background[view_refresh]: radio presets: %v", err) } diff --git a/internal/ingest/packet.go b/internal/ingest/packet.go index b51c99e..eb90db1 100644 --- a/internal/ingest/packet.go +++ b/internal/ingest/packet.go @@ -54,6 +54,7 @@ type InsertObservationParams struct { BandwidthKHz float32 CodingRate int16 SourceBroker string + PayloadType int16 } // RadioSettings holds the radio configuration for an observer, populated from @@ -770,6 +771,7 @@ func (w *Worker) handlePacket(ctx context.Context, iata, pubkeyHex string, raw [ BandwidthKHz: radio.BWKHz, CodingRate: radio.CR, SourceBroker: w.cfg.BrokerName, + PayloadType: int16(packet.PayloadType()), } inserted, err := w.db.InsertObservation(ctx, oParams) if err != nil {