From d3c330e8fb9d3ebf384334ae7ab2286ea103d9eb Mon Sep 17 00:00:00 2001 From: n30nex Date: Tue, 29 Sep 2026 18:27:40 -0400 Subject: [PATCH] fix: address maintainer review for #169 --- README.md | 9 +- .../040_observer_unknown_activity.sql | 279 ------------------ db/observer_metrics_integration_test.go | 7 +- db/observer_window_test.go | 55 ++++ db/observers.go | 3 +- db/observers_test.go | 2 +- db/queries/observer_activity_summary.sql | 13 - db/queries/queries.sql | 14 + db/sqlc/models.go | 2 +- db/sqlc/observer_activity_summary.sql.go | 52 ---- db/sqlc/queries.sql.go | 39 +++ docs/docs.go | 3 +- docs/swagger.json | 3 +- docs/swagger.yaml | 4 +- internal/api/observers.go | 8 +- internal/cache/observer_window_test.go | 47 +++ internal/cache/reader.go | 3 + 17 files changed, 183 insertions(+), 360 deletions(-) delete mode 100644 db/migrations/040_observer_unknown_activity.sql create mode 100644 db/observer_window_test.go delete mode 100644 db/queries/observer_activity_summary.sql delete mode 100644 db/sqlc/observer_activity_summary.sql.go create mode 100644 internal/cache/observer_window_test.go diff --git a/README.md b/README.md index 5f539b5..773cfd9 100644 --- a/README.md +++ b/README.md @@ -553,11 +553,11 @@ new packet deletions. A failed archive leaves its entire raw batch intact. ### Observer monitoring metrics -Observer activity returns complete buckets in `[windowStart, windowEnd)`, plus +Observer activity returns buckets in `[windowStart, windowEnd)`, including the current partial bucket for live requests, plus `generatedAt`, `source` (`raw` or `hourly`) and `summary`. `recordedPackets` is the sum of stored observations in those buckets; repeated broker delivery of the same retained packet/observer pair counts once. Unknown payload types appear as -`-1` rather than disappearing. `lastCompleteHour` uses the previous complete UTC +`-1` rather than disappearing. Freshness fields are measured at `generatedAt` even for historical `until` requests; `recordedPackets` alone follows the selected window. `lastCompleteHour` uses the previous complete UTC hour and includes its own start/end; `latestRecordedAt` is the latest retained reception timestamp. Missing records do not prove downtime. The optional `until` (epoch milliseconds, within the last 30 days) aligns two observers' charts. @@ -566,5 +566,6 @@ The existing observer `observationCount` remains a legacy cumulative presence counter for compatibility, including status/neighbour events. It is not a period packet total. Broker presence and packet-arrival timestamps are now updated separately; this cannot reconstruct previously overwritten timestamps. -Migration 040 repairs archived unknown-type counts from the all-payload observer -rollup; radio samples already discarded for those legacy rows remain unknown. +The initial analytics archive migration includes unknown-type activity. Existing +development previews that used the earlier draft require a separate operator repair; +radio samples already discarded for those legacy rows cannot be recovered. diff --git a/db/migrations/040_observer_unknown_activity.sql b/db/migrations/040_observer_unknown_activity.sql deleted file mode 100644 index 6afab7e..0000000 --- a/db/migrations/040_observer_unknown_activity.sql +++ /dev/null @@ -1,279 +0,0 @@ --- Copyright 2026 Beacon Contributors --- SPDX-License-Identifier: AGPL-3.0-or-later - --- Unknown legacy payloads are still recorded packets. Restore their archived --- counts from the all-payload observer rollup; their discarded radio samples --- cannot be recovered. Subtract all existing activity (including prior repairs) --- so migration-journal retries do not double-count or erase newer samples. -INSERT INTO analytics_observer_activity_hourly -(observer_id,payload_type,bucket,observations,airtime_n,snr_n,rssi_n) -SELECT totals.observer_id, -1, totals.bucket, totals.n-COALESCE(activity.n,0),0,0,0 -FROM (SELECT observer_id,bucket,SUM(observation_count)::bigint AS n - FROM analytics_top_observers_by_iata GROUP BY observer_id,bucket) totals -LEFT JOIN (SELECT observer_id,bucket,SUM(observations)::bigint AS n - FROM analytics_observer_activity_hourly GROUP BY observer_id,bucket) activity -USING (observer_id,bucket) -WHERE totals.n>COALESCE(activity.n,0) -ON CONFLICT (observer_id,payload_type,bucket) DO UPDATE -SET observations=analytics_observer_activity_hourly.observations+EXCLUDED.observations; - -DROP MATERIALIZED VIEW mv_observer_activity_hourly; -CREATE OR REPLACE VIEW analytics_live_observer_activity_hourly AS -SELECT - observer_id, - COALESCE(payload_type, -1)::smallint AS payload_type, - date_trunc('hour', heard_at, 'UTC')::timestamptz AS bucket, - COUNT(*)::bigint AS observations, - SUM(airtime_ms)::real AS airtime_ms, - COUNT(airtime_ms)::bigint AS airtime_n, - SUM(snr) FILTER (WHERE NOT (COALESCE(rssi, 0) = 0 AND COALESCE(snr, 0) = 0))::real AS snr_sum, - COUNT(snr) FILTER (WHERE NOT (COALESCE(rssi, 0) = 0 AND COALESCE(snr, 0) = 0))::bigint AS snr_n, - MIN(snr) FILTER (WHERE NOT (COALESCE(rssi, 0) = 0 AND COALESCE(snr, 0) = 0))::real AS snr_min, - SUM(rssi) FILTER (WHERE NOT (COALESCE(rssi, 0) = 0 AND COALESCE(snr, 0) = 0))::bigint AS rssi_sum, - COUNT(rssi) FILTER (WHERE NOT (COALESCE(rssi, 0) = 0 AND COALESCE(snr, 0) = 0))::bigint AS rssi_n -FROM packet_observations -WHERE heard_at > NOW() - INTERVAL '30 days' -GROUP BY observer_id, COALESCE(payload_type, -1), date_trunc('hour', heard_at, 'UTC'); - -CREATE MATERIALIZED VIEW mv_observer_activity_hourly AS -WITH combined AS ( - SELECT observer_id, payload_type, bucket, observations, airtime_ms, airtime_n, snr_sum, snr_n, snr_min, rssi_sum, rssi_n FROM analytics_live_observer_activity_hourly - UNION ALL - SELECT observer_id, payload_type, bucket, observations, airtime_ms, airtime_n, snr_sum, snr_n, snr_min, rssi_sum, rssi_n FROM analytics_observer_activity_hourly - WHERE bucket >= date_trunc('hour', NOW(), 'UTC') - INTERVAL '30 days' -) -SELECT observer_id, payload_type, bucket, SUM(observations)::bigint AS observations, SUM(airtime_ms)::real AS airtime_ms, SUM(airtime_n)::bigint AS airtime_n, SUM(snr_sum)::real AS snr_sum, SUM(snr_n)::bigint AS snr_n, MIN(snr_min)::real AS snr_min, SUM(rssi_sum)::bigint AS rssi_sum, SUM(rssi_n)::bigint AS rssi_n -FROM combined GROUP BY observer_id,payload_type,bucket; - -CREATE UNIQUE INDEX idx_analytics_observer_activity_hourly_view ON mv_observer_activity_hourly (observer_id,payload_type,bucket); - - - --- Preserve the bounded atomic archive, including unknown payload activity. -CREATE OR REPLACE FUNCTION archive_delete_packets(cutoff timestamptz, batch_size integer) -RETURNS bigint LANGUAGE plpgsql VOLATILE AS $$ -DECLARE hashes bytea[]; deleted bigint; -BEGIN - IF batch_size < 1 THEN RAISE EXCEPTION 'batch_size must be positive'; END IF; - SELECT array_agg(p.packet_hash) INTO hashes FROM ( - SELECT ep.packet_hash FROM packets ep - WHERE ep.last_heard_at < cutoff - ORDER BY ep.last_heard_at, ep.packet_hash - LIMIT batch_size FOR UPDATE OF ep SKIP LOCKED - ) p; - - WITH expired_observations AS MATERIALIZED ( - SELECT po.* FROM packet_observations po WHERE po.packet_hash = ANY(hashes) - ), -archived_hourly_iata_stats AS ( - INSERT INTO analytics_hourly_iata_stats (iata, hour, observation_count, unique_packets) - SELECT iata, hour, observation_count, unique_packets FROM ( -SELECT - iata, - date_trunc('hour', heard_at, 'UTC')::timestamptz AS hour, - COUNT(*) AS observation_count, - COUNT(DISTINCT packet_hash) AS unique_packets -FROM expired_observations -WHERE heard_at >= date_trunc('hour', NOW(), 'UTC') - INTERVAL '30 days' -GROUP BY iata, date_trunc('hour', heard_at, 'UTC') - ) batch - ON CONFLICT (iata,hour) DO UPDATE SET - observation_count = analytics_hourly_iata_stats.observation_count + EXCLUDED.observation_count, - unique_packets = analytics_hourly_iata_stats.unique_packets + EXCLUDED.unique_packets -), -archived_payload_breakdown_by_iata AS ( - INSERT INTO analytics_payload_breakdown_by_iata (iata, payload_type, bucket, count) - SELECT iata, payload_type, bucket, count FROM ( -SELECT - iata, - payload_type, - date_trunc('hour', heard_at, 'UTC')::timestamptz AS bucket, - COUNT(*) AS count -FROM expired_observations -WHERE heard_at >= date_trunc('hour', NOW(), 'UTC') - INTERVAL '30 days' - AND payload_type IS NOT NULL -GROUP BY iata, payload_type, date_trunc('hour', heard_at, 'UTC') - ) batch - ON CONFLICT (iata,payload_type,bucket) DO UPDATE SET - count = analytics_payload_breakdown_by_iata.count + EXCLUDED.count -), -archived_top_observers_by_iata AS ( - INSERT INTO analytics_top_observers_by_iata (iata, observer_id, bucket, observation_count, display_name, observer_type) - SELECT iata, observer_id, bucket, observation_count, display_name, observer_type FROM ( -SELECT - po.iata, - po.observer_id, - o.display_name, - o.observer_type, - date_trunc('hour', po.heard_at, 'UTC')::timestamptz AS bucket, - COUNT(*) AS observation_count -FROM expired_observations po -JOIN observers o ON o.id = po.observer_id -WHERE po.heard_at >= date_trunc('hour', NOW(), 'UTC') - INTERVAL '30 days' -GROUP BY po.iata, po.observer_id, o.display_name, o.observer_type, date_trunc('hour', po.heard_at, 'UTC') - ) batch - ON CONFLICT (iata,observer_id,bucket) DO UPDATE SET - observation_count = analytics_top_observers_by_iata.observation_count + EXCLUDED.observation_count, - display_name = EXCLUDED.display_name, - observer_type = EXCLUDED.observer_type -), -archived_top_talkers_by_iata AS ( - INSERT INTO analytics_top_talkers_by_iata (iata, sender_name, bucket, message_count, last_sent) - SELECT iata, sender_name, bucket, message_count, last_sent FROM ( -SELECT - po.iata, - cm.sender_name, - date_trunc('hour', cm.sent_at, 'UTC')::timestamptz AS bucket, - COUNT(DISTINCT cm.id) AS message_count, - MAX(cm.sent_at) AS last_sent -FROM channel_messages cm -JOIN expired_observations po ON po.packet_hash = cm.packet_hash -WHERE cm.sender_name IS NOT NULL - AND cm.sent_at >= date_trunc('hour', NOW(), 'UTC') - INTERVAL '30 days' -GROUP BY po.iata, cm.sender_name, date_trunc('hour', cm.sent_at, 'UTC') - ) batch - ON CONFLICT (iata,sender_name,bucket) DO UPDATE SET - message_count = analytics_top_talkers_by_iata.message_count + EXCLUDED.message_count, - last_sent = GREATEST(analytics_top_talkers_by_iata.last_sent, EXCLUDED.last_sent) -), -archived_top_advertisers_by_iata AS ( - INSERT INTO analytics_top_advertisers_by_iata (iata, node_id, bucket, advert_count, flood_advert_count, direct_advert_count, last_heard, name, node_type) - SELECT iata, node_id, bucket, advert_count, flood_advert_count, direct_advert_count, last_heard, name, node_type FROM ( -SELECT - po.iata, - n.id AS node_id, - n.name, - n.node_type, - date_trunc('hour', po.heard_at, 'UTC')::timestamptz AS bucket, - COUNT(DISTINCT p.packet_hash) AS advert_count, - COUNT(DISTINCT p.packet_hash) FILTER (WHERE p.route_type IN (0, 1)) AS flood_advert_count, - COUNT(DISTINCT p.packet_hash) FILTER (WHERE p.route_type IN (2, 3)) AS direct_advert_count, - MAX(po.heard_at) AS last_heard -FROM packets p -JOIN expired_observations po ON po.packet_hash = p.packet_hash -JOIN nodes n ON n.public_key = p.origin_pubkey -WHERE p.payload_type = 4 -- ADVERT - AND po.heard_at >= date_trunc('hour', NOW(), 'UTC') - INTERVAL '30 days' -GROUP BY po.iata, n.id, n.name, n.node_type, date_trunc('hour', po.heard_at, 'UTC') - ) batch - ON CONFLICT (iata,node_id,bucket) DO UPDATE SET - advert_count = analytics_top_advertisers_by_iata.advert_count + EXCLUDED.advert_count, - flood_advert_count = analytics_top_advertisers_by_iata.flood_advert_count + EXCLUDED.flood_advert_count, - direct_advert_count = analytics_top_advertisers_by_iata.direct_advert_count + EXCLUDED.direct_advert_count, - last_heard = GREATEST(analytics_top_advertisers_by_iata.last_heard, EXCLUDED.last_heard), - name = EXCLUDED.name, - node_type = EXCLUDED.node_type -), -archived_observer_activity_hourly AS ( - INSERT INTO analytics_observer_activity_hourly (observer_id, payload_type, bucket, observations, airtime_ms, airtime_n, snr_sum, snr_n, snr_min, rssi_sum, rssi_n) - SELECT observer_id, payload_type, bucket, observations, airtime_ms, airtime_n, snr_sum, snr_n, snr_min, rssi_sum, rssi_n FROM ( -SELECT - observer_id, - COALESCE(payload_type, -1)::smallint AS payload_type, - date_trunc('hour', heard_at, 'UTC')::timestamptz AS bucket, - COUNT(*)::bigint AS observations, - SUM(airtime_ms)::real AS airtime_ms, - COUNT(airtime_ms)::bigint AS airtime_n, - SUM(snr) FILTER (WHERE NOT (COALESCE(rssi, 0) = 0 AND COALESCE(snr, 0) = 0))::real AS snr_sum, - COUNT(snr) FILTER (WHERE NOT (COALESCE(rssi, 0) = 0 AND COALESCE(snr, 0) = 0))::bigint AS snr_n, - MIN(snr) FILTER (WHERE NOT (COALESCE(rssi, 0) = 0 AND COALESCE(snr, 0) = 0))::real AS snr_min, - SUM(rssi) FILTER (WHERE NOT (COALESCE(rssi, 0) = 0 AND COALESCE(snr, 0) = 0))::bigint AS rssi_sum, - COUNT(rssi) FILTER (WHERE NOT (COALESCE(rssi, 0) = 0 AND COALESCE(snr, 0) = 0))::bigint AS rssi_n -FROM expired_observations -WHERE heard_at >= date_trunc('hour', NOW(), 'UTC') - INTERVAL '30 days' -GROUP BY observer_id, COALESCE(payload_type, -1), date_trunc('hour', heard_at, 'UTC') - ) batch - ON CONFLICT (observer_id,payload_type,bucket) DO UPDATE SET - observations = analytics_observer_activity_hourly.observations + EXCLUDED.observations, - airtime_ms = CASE WHEN analytics_observer_activity_hourly.airtime_ms IS NULL AND EXCLUDED.airtime_ms IS NULL THEN NULL ELSE COALESCE(analytics_observer_activity_hourly.airtime_ms, 0) + COALESCE(EXCLUDED.airtime_ms, 0) END, - airtime_n = analytics_observer_activity_hourly.airtime_n + EXCLUDED.airtime_n, - snr_sum = CASE WHEN analytics_observer_activity_hourly.snr_sum IS NULL AND EXCLUDED.snr_sum IS NULL THEN NULL ELSE COALESCE(analytics_observer_activity_hourly.snr_sum, 0) + COALESCE(EXCLUDED.snr_sum, 0) END, - snr_n = analytics_observer_activity_hourly.snr_n + EXCLUDED.snr_n, - snr_min = LEAST(analytics_observer_activity_hourly.snr_min, EXCLUDED.snr_min), - rssi_sum = CASE WHEN analytics_observer_activity_hourly.rssi_sum IS NULL AND EXCLUDED.rssi_sum IS NULL THEN NULL ELSE COALESCE(analytics_observer_activity_hourly.rssi_sum, 0) + COALESCE(EXCLUDED.rssi_sum, 0) END, - rssi_n = analytics_observer_activity_hourly.rssi_n + EXCLUDED.rssi_n -), -archived_signal_stats_hourly AS ( - INSERT INTO analytics_signal_stats_hourly (iata, hour, kind, snr_bin, rssi_bin, receptions, snr_samples, snr_sum, rssi_samples, rssi_sum) - SELECT iata, hour, kind, snr_bin, rssi_bin, receptions, snr_samples, snr_sum, rssi_samples, rssi_sum FROM ( -WITH samples AS ( - SELECT iata, date_trunc('hour', heard_at, 'UTC') AS hour, - CASE WHEN NOT (COALESCE(rssi, 0) = 0 AND COALESCE(snr, 0) = 0) - AND snr > '-Infinity'::real AND snr < 'Infinity'::real - THEN snr::double precision END AS snr, - CASE WHEN NOT (COALESCE(rssi, 0) = 0 AND COALESCE(snr, 0) = 0) - THEN rssi::double precision END AS rssi - FROM expired_observations - WHERE heard_at >= date_trunc('hour', NOW(), 'UTC') - INTERVAL '720 hours' - -), binned AS ( - SELECT *, width_bucket(snr, -30, 30, 12) AS snr_bin, - width_bucket(rssi, -140, 0, 14) AS rssi_bin - FROM samples -) -SELECT iata, hour, grouping(snr_bin, rssi_bin)::integer AS kind, - COALESCE(snr_bin, -1)::integer AS snr_bin, - COALESCE(rssi_bin, -1)::integer AS rssi_bin, - count(*)::bigint AS receptions, - count(snr)::bigint AS snr_samples, - COALESCE(sum(snr), 0)::double precision AS snr_sum, - count(rssi)::bigint AS rssi_samples, - COALESCE(sum(rssi), 0)::double precision AS rssi_sum -FROM binned -GROUP BY GROUPING SETS ((iata, hour), (iata, hour, snr_bin), (iata, hour, rssi_bin)) - ) batch - ON CONFLICT (iata,hour,kind,snr_bin,rssi_bin) DO UPDATE SET - receptions = analytics_signal_stats_hourly.receptions + EXCLUDED.receptions, - snr_samples = analytics_signal_stats_hourly.snr_samples + EXCLUDED.snr_samples, - snr_sum = analytics_signal_stats_hourly.snr_sum + EXCLUDED.snr_sum, - rssi_samples = analytics_signal_stats_hourly.rssi_samples + EXCLUDED.rssi_samples, - rssi_sum = analytics_signal_stats_hourly.rssi_sum + EXCLUDED.rssi_sum -), -archived_path_stats_hourly AS ( - INSERT INTO analytics_path_stats_hourly (iata, hour, category, hash_bytes, entries, receptions) - SELECT iata, hour, category, hash_bytes, entries, receptions FROM ( -WITH classified AS ( - SELECT iata, heard_at, hash_size, hop_count, - CASE WHEN payload_type = 9 THEN 2 - -- Match meshcore-go IsValidPathLen (1/2/3-byte hashes, max 64 path bytes). - WHEN payload_type IS NULL OR payload_type NOT BETWEEN 0 AND 15 - OR NOT (path_length_byte BETWEEN 0 AND 191 - AND hash_size BETWEEN 1 AND 3 AND hop_count BETWEEN 0 AND 63 - AND hash_size = (path_length_byte >> 6) + 1 - AND hop_count = (path_length_byte & 63) - AND hash_size::integer * hop_count::integer <= 64 - AND COALESCE(octet_length(path_bytes), 0) = hash_size::integer * hop_count::integer) - THEN 3 - WHEN hop_count = 0 THEN 1 - ELSE 0 END::integer AS category - FROM expired_observations - WHERE heard_at >= date_trunc('hour', NOW(), 'UTC') - INTERVAL '720 hours' - -), buckets AS ( - SELECT iata, date_trunc('hour', heard_at, 'UTC') AS hour, category, - CASE WHEN category = 0 THEN hash_size ELSE 0 END::integer AS hash_bytes, - CASE WHEN category = 0 THEN hop_count ELSE 0 END::integer AS entries - FROM classified -) -SELECT iata, hour, category, hash_bytes, entries, count(*)::bigint AS receptions -FROM buckets GROUP BY iata, hour, category, hash_bytes, entries - ) batch - ON CONFLICT (iata,hour,category,hash_bytes,entries) DO UPDATE SET - receptions = analytics_path_stats_hourly.receptions + EXCLUDED.receptions -) - DELETE FROM packets WHERE packet_hash = ANY(hashes); - GET DIAGNOSTICS deleted = ROW_COUNT; - - -- Once per completed cleanup, including when there were no raw packets to delete. - IF deleted < batch_size THEN - DELETE FROM analytics_hourly_iata_stats WHERE hour < date_trunc('hour', NOW(), 'UTC') - INTERVAL '30 days'; - DELETE FROM analytics_payload_breakdown_by_iata WHERE bucket < date_trunc('hour', NOW(), 'UTC') - INTERVAL '30 days'; - DELETE FROM analytics_top_observers_by_iata WHERE bucket < date_trunc('hour', NOW(), 'UTC') - INTERVAL '30 days'; - DELETE FROM analytics_top_talkers_by_iata WHERE bucket < date_trunc('hour', NOW(), 'UTC') - INTERVAL '30 days'; - DELETE FROM analytics_top_advertisers_by_iata WHERE bucket < date_trunc('hour', NOW(), 'UTC') - INTERVAL '30 days'; - DELETE FROM analytics_observer_activity_hourly WHERE bucket < date_trunc('hour', NOW(), 'UTC') - INTERVAL '30 days'; - DELETE FROM analytics_signal_stats_hourly WHERE hour < date_trunc('hour', NOW(), 'UTC') - INTERVAL '30 days'; - DELETE FROM analytics_path_stats_hourly WHERE hour < date_trunc('hour', NOW(), 'UTC') - INTERVAL '30 days'; - END IF; - RETURN deleted; -END $$; diff --git a/db/observer_metrics_integration_test.go b/db/observer_metrics_integration_test.go index 6266dff..91dbeca 100644 --- a/db/observer_metrics_integration_test.go +++ b/db/observer_metrics_integration_test.go @@ -41,9 +41,10 @@ func TestObserverMetricsPostgres(t *testing.T) { if err := store.DeleteOldPackets(ctx, time.Now().Add(-72*time.Hour)); err != nil { t.Fatal(err) } - // Migration repairs the two unknown counts that 039 did not archive. - applyStatsMigration(t, ctx, tx, "040_observer_unknown_activity.sql") - applyStatsMigration(t, ctx, tx, "040_observer_unknown_activity.sql") + // The initial archive migration retains unknown payloads without a second rebuild. + if err := store.RefreshObserverActivity(ctx); err != nil { + t.Fatal(err) + } until := time.Now().UTC().Truncate(time.Hour) activity, err := store.GetObserverActivity(ctx, id, 7*24*time.Hour, time.Hour, until) if err != nil { diff --git a/db/observer_window_test.go b/db/observer_window_test.go new file mode 100644 index 0000000..205d1ef --- /dev/null +++ b/db/observer_window_test.go @@ -0,0 +1,55 @@ +// Copyright 2026 Beacon Contributors +// SPDX-License-Identifier: AGPL-3.0-or-later + +package db + +import ( + "context" + "testing" + "time" + + sqlc "github.com/MeshCore-Beacon/beacon-server/db/sqlc" + mockdb "github.com/MeshCore-Beacon/beacon-server/db/sqlc/mock" + "github.com/google/uuid" + "go.uber.org/mock/gomock" +) + +func TestObserverActivityLiveAndFixedWindow(t *testing.T) { + for _, interval := range []time.Duration{15 * time.Minute, time.Hour, 6 * time.Hour} { + for _, fixed := range []bool{false, true} { + t.Run(interval.String()+"/"+map[bool]string{false: "live", true: "fixed"}[fixed], func(t *testing.T) { + mock := mockdb.NewMockQuerier(gomock.NewController(t)) + id := uuid.New() + mock.EXPECT().GetObserverByID(gomock.Any(), id).Return(sqlc.Observer{}, nil) + mock.EXPECT().GetObserverActivityLiveSummary(gomock.Any(), gomock.Any()).Return(sqlc.GetObserverActivityLiveSummaryRow{}, nil) + if interval < time.Hour { + mock.EXPECT().GetObserverActivityRaw(gomock.Any(), gomock.Any()).Return(nil, nil) + mock.EXPECT().GetObserverActivityRawPayloadTypes(gomock.Any(), gomock.Any()).Return(nil, nil) + } else { + mock.EXPECT().GetObserverActivityHourly(gomock.Any(), gomock.Any()).Return(nil, nil) + mock.EXPECT().GetObserverActivityHourlyPayloadTypes(gomock.Any(), gomock.Any()).Return(nil, nil) + } + before := time.Now().UTC() + var until time.Time + if fixed { + until = before.Add(-24*time.Hour + 123*time.Millisecond) + } + got, err := (&Store{q: mock}).GetObserverActivity(context.Background(), id, 48*time.Hour, interval, until) + if err != nil { + t.Fatal(err) + } + if fixed { + if got.WindowEnd != until.Truncate(interval).UnixMilli() { + t.Fatalf("fixed end was not aligned: %d", got.WindowEnd) + } + } else if got.WindowEnd < before.UnixMilli() || got.WindowEnd > time.Now().UnixMilli() { + t.Fatalf("live end dropped the partial bucket: %d", got.WindowEnd) + } + // Freshness is always live, even when the activity window is historical. + if got.Summary.LastCompleteHourEnd != time.UnixMilli(got.GeneratedAt).UTC().Truncate(time.Hour).UnixMilli() { + t.Fatal("freshness was anchored to the historical activity window") + } + }) + } + } +} diff --git a/db/observers.go b/db/observers.go index 09a377a..36316db 100644 --- a/db/observers.go +++ b/db/observers.go @@ -239,8 +239,9 @@ func (s *Store) GetObserverActivity(ctx context.Context, observerID uuid.UUID, w now := time.Now().UTC() if until.IsZero() { until = now + } else { + until = until.UTC().Truncate(interval) } - until = until.UTC().Truncate(interval) since := until.Add(-window) if !since.Equal(since.Truncate(interval)) { since = since.Truncate(interval).Add(interval) diff --git a/db/observers_test.go b/db/observers_test.go index 31b562e..44ed23c 100644 --- a/db/observers_test.go +++ b/db/observers_test.go @@ -919,7 +919,7 @@ func TestGetObserverActivity_SinceAlignedToInterval(t *testing.T) { if gotSince.Before(before) { t.Errorf("since %s is before the window start %s", gotSince, before) } - if !gotSince.Before(after.Add(interval)) { + if gotSince.After(after.Add(interval)) { t.Errorf("since %s is more than one interval past the window start %s", gotSince, after) } if !gotInterval.Valid || gotInterval.Microseconds != interval.Microseconds() { diff --git a/db/queries/observer_activity_summary.sql b/db/queries/observer_activity_summary.sql deleted file mode 100644 index 55b5aed..0000000 --- a/db/queries/observer_activity_summary.sql +++ /dev/null @@ -1,13 +0,0 @@ --- name: GetObserverActivityLiveSummary :one --- Two indexed ranges, bounded to one observer; no legacy presence counters. -WITH latest AS ( - SELECT heard_at FROM packet_observations - WHERE observer_id = @observer_id::uuid AND heard_at <= @generated_at::timestamptz - ORDER BY heard_at DESC LIMIT 1 -), hourly AS ( - SELECT COUNT(*)::bigint AS n FROM packet_observations - WHERE observer_id = @observer_id::uuid - AND heard_at >= @hour_start::timestamptz AND heard_at < @hour_end::timestamptz -) -SELECT (SELECT heard_at FROM latest)::timestamptz AS latest_recorded_at, - hourly.n AS last_complete_hour FROM hourly; diff --git a/db/queries/queries.sql b/db/queries/queries.sql index aaf0be8..05ceacd 100644 --- a/db/queries/queries.sql +++ b/db/queries/queries.sql @@ -1572,3 +1572,17 @@ WITH target AS MATERIALIZED ( ) SELECT EXISTS(SELECT 1 FROM target) AS found, EXISTS(SELECT 1 FROM changed) AS deactivated; + +-- name: GetObserverActivityLiveSummary :one +-- Two indexed ranges, bounded to one observer; no legacy presence counters. +WITH latest AS ( + SELECT heard_at FROM packet_observations + WHERE observer_id = @observer_id::uuid AND heard_at <= @generated_at::timestamptz + ORDER BY heard_at DESC LIMIT 1 +), hourly AS ( + SELECT COUNT(*)::bigint AS n FROM packet_observations + WHERE observer_id = @observer_id::uuid + AND heard_at >= @hour_start::timestamptz AND heard_at < @hour_end::timestamptz +) +SELECT (SELECT heard_at FROM latest)::timestamptz AS latest_recorded_at, + hourly.n AS last_complete_hour FROM hourly; diff --git a/db/sqlc/models.go b/db/sqlc/models.go index 8a24d53..378fb60 100644 --- a/db/sqlc/models.go +++ b/db/sqlc/models.go @@ -32,7 +32,7 @@ type AnalyticsLiveHourlyIataStat struct { type AnalyticsLiveObserverActivityHourly struct { ObserverID uuid.UUID `json:"observer_id"` - PayloadType int16 `json:"payload_type"` + PayloadType *int16 `json:"payload_type"` Bucket pgtype.Timestamptz `json:"bucket"` Observations int64 `json:"observations"` AirtimeMs float32 `json:"airtime_ms"` diff --git a/db/sqlc/observer_activity_summary.sql.go b/db/sqlc/observer_activity_summary.sql.go deleted file mode 100644 index c6c97ac..0000000 --- a/db/sqlc/observer_activity_summary.sql.go +++ /dev/null @@ -1,52 +0,0 @@ -// Code generated by sqlc. DO NOT EDIT. -// versions: -// sqlc v1.31.1 -// source: observer_activity_summary.sql - -package db - -import ( - "context" - - "github.com/google/uuid" - "github.com/jackc/pgx/v5/pgtype" -) - -const getObserverActivityLiveSummary = `-- name: GetObserverActivityLiveSummary :one -WITH latest AS ( - SELECT heard_at FROM packet_observations - WHERE observer_id = $1::uuid AND heard_at <= $2::timestamptz - ORDER BY heard_at DESC LIMIT 1 -), hourly AS ( - SELECT COUNT(*)::bigint AS n FROM packet_observations - WHERE observer_id = $1::uuid - AND heard_at >= $3::timestamptz AND heard_at < $4::timestamptz -) -SELECT (SELECT heard_at FROM latest)::timestamptz AS latest_recorded_at, - hourly.n AS last_complete_hour FROM hourly -` - -type GetObserverActivityLiveSummaryParams struct { - ObserverID uuid.UUID `json:"observer_id"` - GeneratedAt pgtype.Timestamptz `json:"generated_at"` - HourStart pgtype.Timestamptz `json:"hour_start"` - HourEnd pgtype.Timestamptz `json:"hour_end"` -} - -type GetObserverActivityLiveSummaryRow struct { - LatestRecordedAt pgtype.Timestamptz `json:"latest_recorded_at"` - LastCompleteHour int64 `json:"last_complete_hour"` -} - -// Two indexed ranges, bounded to one observer; no legacy presence counters. -func (q *Queries) GetObserverActivityLiveSummary(ctx context.Context, arg GetObserverActivityLiveSummaryParams) (GetObserverActivityLiveSummaryRow, error) { - row := q.db.QueryRow(ctx, getObserverActivityLiveSummary, - arg.ObserverID, - arg.GeneratedAt, - arg.HourStart, - arg.HourEnd, - ) - var i GetObserverActivityLiveSummaryRow - err := row.Scan(&i.LatestRecordedAt, &i.LastCompleteHour) - return i, err -} diff --git a/db/sqlc/queries.sql.go b/db/sqlc/queries.sql.go index 3dd6de8..489094c 100644 --- a/db/sqlc/queries.sql.go +++ b/db/sqlc/queries.sql.go @@ -749,6 +749,45 @@ func (q *Queries) GetObserverActivityHourlyPayloadTypes(ctx context.Context, arg return items, nil } +const getObserverActivityLiveSummary = `-- name: GetObserverActivityLiveSummary :one +WITH latest AS ( + SELECT heard_at FROM packet_observations + WHERE observer_id = $1::uuid AND heard_at <= $2::timestamptz + ORDER BY heard_at DESC LIMIT 1 +), hourly AS ( + SELECT COUNT(*)::bigint AS n FROM packet_observations + WHERE observer_id = $1::uuid + AND heard_at >= $3::timestamptz AND heard_at < $4::timestamptz +) +SELECT (SELECT heard_at FROM latest)::timestamptz AS latest_recorded_at, + hourly.n AS last_complete_hour FROM hourly +` + +type GetObserverActivityLiveSummaryParams struct { + ObserverID uuid.UUID `json:"observer_id"` + GeneratedAt pgtype.Timestamptz `json:"generated_at"` + HourStart pgtype.Timestamptz `json:"hour_start"` + HourEnd pgtype.Timestamptz `json:"hour_end"` +} + +type GetObserverActivityLiveSummaryRow struct { + LatestRecordedAt pgtype.Timestamptz `json:"latest_recorded_at"` + LastCompleteHour int64 `json:"last_complete_hour"` +} + +// Two indexed ranges, bounded to one observer; no legacy presence counters. +func (q *Queries) GetObserverActivityLiveSummary(ctx context.Context, arg GetObserverActivityLiveSummaryParams) (GetObserverActivityLiveSummaryRow, error) { + row := q.db.QueryRow(ctx, getObserverActivityLiveSummary, + arg.ObserverID, + arg.GeneratedAt, + arg.HourStart, + arg.HourEnd, + ) + var i GetObserverActivityLiveSummaryRow + err := row.Scan(&i.LatestRecordedAt, &i.LastCompleteHour) + return i, err +} + const getObserverActivityRaw = `-- name: GetObserverActivityRaw :many SELECT date_bin($3::interval, heard_at, TIMESTAMPTZ 'epoch')::timestamptz AS bucket, diff --git a/docs/docs.go b/docs/docs.go index bc88fbf..1b27101 100644 --- a/docs/docs.go +++ b/docs/docs.go @@ -4019,7 +4019,7 @@ const docTemplate = `{ "$ref": "#/definitions/github_com_MeshCore-Beacon_beacon-server_internal_api.ObserverActivitySummary" }, "windowEnd": { - "description": "exclusive complete-bucket end, epoch ms", + "description": "exclusive end, epoch ms; live requests include the current partial bucket", "type": "integer" }, "windowStart": { @@ -4088,6 +4088,7 @@ const docTemplate = `{ "type": "integer" }, "recordedPackets": { + "description": "stored observations within windowStart/windowEnd", "type": "integer" } } diff --git a/docs/swagger.json b/docs/swagger.json index 75a8e5a..52df28f 100644 --- a/docs/swagger.json +++ b/docs/swagger.json @@ -4017,7 +4017,7 @@ "$ref": "#/definitions/github_com_MeshCore-Beacon_beacon-server_internal_api.ObserverActivitySummary" }, "windowEnd": { - "description": "exclusive complete-bucket end, epoch ms", + "description": "exclusive end, epoch ms; live requests include the current partial bucket", "type": "integer" }, "windowStart": { @@ -4086,6 +4086,7 @@ "type": "integer" }, "recordedPackets": { + "description": "stored observations within windowStart/windowEnd", "type": "integer" } } diff --git a/docs/swagger.yaml b/docs/swagger.yaml index cab6df7..4fda3a4 100644 --- a/docs/swagger.yaml +++ b/docs/swagger.yaml @@ -584,7 +584,8 @@ definitions: summary: $ref: '#/definitions/github_com_MeshCore-Beacon_beacon-server_internal_api.ObserverActivitySummary' windowEnd: - description: exclusive complete-bucket end, epoch ms + description: exclusive end, epoch ms; live requests include the current partial + bucket type: integer windowStart: description: inclusive complete-bucket start, epoch ms @@ -630,6 +631,7 @@ definitions: latestRecordedAt: type: integer recordedPackets: + description: stored observations within windowStart/windowEnd type: integer type: object github_com_MeshCore-Beacon_beacon-server_internal_api.ObserverBroker: diff --git a/internal/api/observers.go b/internal/api/observers.go index d16fb13..2da680c 100644 --- a/internal/api/observers.go +++ b/internal/api/observers.go @@ -88,19 +88,21 @@ type ObserverActivityPoint struct { RSSIAvg *float32 `json:"rssiAvg"` } -// ObserverActivity is the per-observer heard-activity response. // ObserverActivitySummary describes stored packet records, never MQTT presence events. +// Freshness and lastCompleteHour are measured at generatedAt, even with an explicit until; +// only recordedPackets follows the selected activity window. type ObserverActivitySummary struct { - RecordedPackets int64 `json:"recordedPackets"` + RecordedPackets int64 `json:"recordedPackets"` // stored observations within windowStart/windowEnd LastCompleteHour int64 `json:"lastCompleteHour"` LastCompleteHourStart int64 `json:"lastCompleteHourStart"` LastCompleteHourEnd int64 `json:"lastCompleteHourEnd"` LatestRecordedAt *int64 `json:"latestRecordedAt"` } +// ObserverActivity is the per-observer heard-activity response. type ObserverActivity struct { WindowStart int64 `json:"windowStart"` // inclusive complete-bucket start, epoch ms - WindowEnd int64 `json:"windowEnd"` // exclusive complete-bucket end, epoch ms + WindowEnd int64 `json:"windowEnd"` // exclusive end, epoch ms; live requests include the current partial bucket GeneratedAt int64 `json:"generatedAt"` // response computation time, not proof of continuous coverage Source string `json:"source"` // raw or hourly; missing records do not prove an outage Summary *ObserverActivitySummary `json:"summary,omitempty"` diff --git a/internal/cache/observer_window_test.go b/internal/cache/observer_window_test.go new file mode 100644 index 0000000..512e580 --- /dev/null +++ b/internal/cache/observer_window_test.go @@ -0,0 +1,47 @@ +// Copyright 2026 Beacon Contributors +// SPDX-License-Identifier: AGPL-3.0-or-later + +package cache + +import ( + "context" + "testing" + "time" + + "github.com/MeshCore-Beacon/beacon-server/internal/api" + "github.com/google/uuid" +) + +type activityWindowReader struct { + stubReader + ends []time.Time +} + +func (r *activityWindowReader) GetObserverActivity(_ context.Context, _ uuid.UUID, _, _ time.Duration, until time.Time) (*api.ObserverActivity, error) { + r.ends = append(r.ends, until) + return &api.ObserverActivity{WindowEnd: until.UnixMilli()}, nil +} + +func TestObserverActivityCacheUsesEffectiveWindow(t *testing.T) { + client, redis := newTestClient(t) + inner := &activityWindowReader{} + reader := &CachedReader{inner: inner, c: client} + id := uuid.New() + anchor := time.Date(2026, 9, 29, 12, 0, 0, 0, time.UTC) + for _, offset := range []time.Duration{time.Second, time.Minute, 59 * time.Minute} { + if _, err := reader.GetObserverActivity(context.Background(), id, 24*time.Hour, time.Hour, anchor.Add(offset)); err != nil { + t.Fatal(err) + } + } + if len(inner.ends) != 1 || !inner.ends[0].Equal(anchor) || len(redis.Keys()) != 1 { + t.Fatalf("equivalent fixed windows missed the cache: %v, %v", inner.ends, redis.Keys()) + } + for _, until := range []time.Time{anchor.Add(time.Hour), {}, {}} { + if _, err := reader.GetObserverActivity(context.Background(), id, 24*time.Hour, time.Hour, until); err != nil { + t.Fatal(err) + } + } + if len(inner.ends) != 3 || !inner.ends[2].IsZero() || len(redis.Keys()) != 3 { + t.Fatalf("distinct/live windows aliased: %v", inner.ends) + } +} diff --git a/internal/cache/reader.go b/internal/cache/reader.go index 2d31ba4..2cec676 100644 --- a/internal/cache/reader.go +++ b/internal/cache/reader.go @@ -373,6 +373,9 @@ func (cr *CachedReader) GetObserverTelemetryBucketed(ctx context.Context, observ // GetObserverActivity implements [api.Reader]. func (cr *CachedReader) GetObserverActivity(ctx context.Context, observerID uuid.UUID, window, interval time.Duration, until time.Time) (*api.ObserverActivity, error) { + if !until.IsZero() { + until = until.UTC().Truncate(interval) + } key := keyObserverActivityPrefix + observerID.String() + ":" + window.String() + ":" + interval.String() + ":" + until.UTC().Format(time.RFC3339Nano) return getOrSet(ctx, cr.c, key, observerActivityTTL, func() (*api.ObserverActivity, error) { return cr.inner.GetObserverActivity(ctx, observerID, window, interval, until)