Observer comparison: read each observer's receptions on their own (#260)

* Read each observer's receptions on their own in the observer comparison

* Tighten the observer comparison plan checks

.github/workflows/ci.yml

@@ -69,7 +69,7 @@ jobs:
       - name: Verify stats and endpoint queries against PostgreSQL 16
         env:
           BEACON_TEST_POSTGRES_DSN: postgres://postgres:backup-ci-only@127.0.0.1:5432/postgres?sslmode=disable
-        run: go test ./db -run '^Test(Signal|Paths|PacketSummaries|PacketEndpointsResolveLive|ObserverMetrics|RouteEvidence|RouteEvidenceIndex|AnalyticsRetention|AnalyticsRetentionConcurrent|DeleteOldPacketsBatches|MeshMapperCatalogue|ChannelMessageScopes|NodeLocationReset|RunMigrations[A-Za-z]*|KnownRoutesIndexSet|PacketsDomainChecks|Reconfirm[A-Za-z]*|BigintCursors|RouteSearchHopOrder|StatsNodeTypesSplit|ListKnownRoutes|ListKnownRoutesMatchesLegacy|KnownRoutesCursorTies|KnownRoutesUnanalyzedRegion|KnownRoutesHopStatementsStayCustom|KnownRoutePaginationTies|RouteRetentionCutoffsExact|DeleteExpiredRoutes|DeleteGraceRoutesSlice|GraceWatermarkHeldByLockedRow|OpenPoolsSettings|PlanRouterClassCUnnamed|APIStoreHopStatementsUnnamed|PgStatStatementsMigrationGuarded|DBStatements|DBStatementsNoExtension|PgStatStatementsMigrationNoCreatableSchema|IndexHealthBaselineOnReindex|UnusedIndexes|PgActivity|RollHourSetsWorkMem|SmallTableIndexSet|NodeReadersAfterIndexDrops|ObserverChannelReadersAfterIndexDrops|UpsertChannelWriteThrough|UpsertChannelHashOnlyWriteThrough|ReadOnlyPoolsRejectWrites|ListPacketsGlobalMatchesLegacy|PacketsGlobalCursorTies|ListPacketsWindowInvertedRows|PacketsLatestObserverTie|NodesLegacyCursorMatchesToday|NodesPageCursorTies|ObserversLegacyCursorMatchesToday|ObserversNextCursorFromLastSeen|ObserverAdvertsPartialIndex|AdvertsWaitForPayloadTypeBackfill|StatsProbeMatchesToday|TraceTagsKeysetMatchesToday|TelemetrySinceMatchesCatchAll|TraceTagPagination|TraceTagPaginationTies|TraceSummarySurvivesMissedHearing|CmiMigrationsIndexSet|InsertChannelMessageSeed[A-Za-z]*|AddChannelMessageIATA|CmiSeedAndAddRace|CmiInitAndBackfill|CmiMessageRangeReconcile|CmiObservationRangeReconcile|CmiBackfillRetries23503|WaitForPreBootWriters|ChannelMessagesByIATAsMatchesLegacy|ChannelMessagesZeroHitRegion|CmiTrustLifecycle|CmiFirstVerification|PacketSitesMigrationsIndexSet|InsertObservationMaintainsPacketSites|InsertObservationResendWritesNothing|PacketSitesRaceMaxMerge|InsertObservationSiteFallback|PacketSitesInitAndBackfill|FillPacketSitesSliceIdempotent|FillPacketSitesSliceRetries23503|PacketSitesReconcile|PacketSitesAudit|PacketSitesAuditTakesNewestHearing|PacketsBySitesMatchesTruth|PacketsBySitesCursorTies|PacketsBySitesManyBogusSites|PacketsBySitesSaturatedPagesFull|PacketsBySitesMatchesLegacy|PacketSitesTrustLifecycle|PacketsBySitesBounded|PacketsBySitesFallbackKeepsPageCursor|SweepPacketSitesFrontsPlan|IndexHealthPacketSitesFreshBaseline|PacketSitesFirstVerification|PacedBackfillSlices|FillSessionsPlanCustom|FillAnalyzeFailureIsIgnored|FillAnalyzeOnlyNeverAnalyzed|HygieneIndexSet|HygieneReadersAfterIndexDrops|KnownRoutesColumns|SequenceUse|SlowPoolIsolation|SlowPoolSize|SequenceUseMixedCaseSchema|ObservationsIndexSet|ReadersAfterIndexDrops)Postgres$' -count=1 -v
+        run: go test ./db -run '^Test(Signal|Paths|PacketSummaries|PacketEndpointsResolveLive|ObserverMetrics|RouteEvidence|RouteEvidenceIndex|AnalyticsRetention|AnalyticsRetentionConcurrent|DeleteOldPacketsBatches|MeshMapperCatalogue|ChannelMessageScopes|NodeLocationReset|RunMigrations[A-Za-z]*|KnownRoutesIndexSet|PacketsDomainChecks|Reconfirm[A-Za-z]*|BigintCursors|RouteSearchHopOrder|StatsNodeTypesSplit|ListKnownRoutes|ListKnownRoutesMatchesLegacy|KnownRoutesCursorTies|KnownRoutesUnanalyzedRegion|KnownRoutesHopStatementsStayCustom|KnownRoutePaginationTies|RouteRetentionCutoffsExact|DeleteExpiredRoutes|DeleteGraceRoutesSlice|GraceWatermarkHeldByLockedRow|OpenPoolsSettings|PlanRouterClassCUnnamed|APIStoreHopStatementsUnnamed|PgStatStatementsMigrationGuarded|DBStatements|DBStatementsNoExtension|PgStatStatementsMigrationNoCreatableSchema|IndexHealthBaselineOnReindex|UnusedIndexes|PgActivity|RollHourSetsWorkMem|SmallTableIndexSet|NodeReadersAfterIndexDrops|ObserverChannelReadersAfterIndexDrops|UpsertChannelWriteThrough|UpsertChannelHashOnlyWriteThrough|ReadOnlyPoolsRejectWrites|ListPacketsGlobalMatchesLegacy|PacketsGlobalCursorTies|ListPacketsWindowInvertedRows|PacketsLatestObserverTie|NodesLegacyCursorMatchesToday|NodesPageCursorTies|ObserversLegacyCursorMatchesToday|ObserversNextCursorFromLastSeen|ObserverAdvertsPartialIndex|ObserverComparisonMatchesToday|AdvertsWaitForPayloadTypeBackfill|StatsProbeMatchesToday|TraceTagsKeysetMatchesToday|TelemetrySinceMatchesCatchAll|TraceTagPagination|TraceTagPaginationTies|TraceSummarySurvivesMissedHearing|CmiMigrationsIndexSet|InsertChannelMessageSeed[A-Za-z]*|AddChannelMessageIATA|CmiSeedAndAddRace|CmiInitAndBackfill|CmiMessageRangeReconcile|CmiObservationRangeReconcile|CmiBackfillRetries23503|WaitForPreBootWriters|ChannelMessagesByIATAsMatchesLegacy|ChannelMessagesZeroHitRegion|CmiTrustLifecycle|CmiFirstVerification|PacketSitesMigrationsIndexSet|InsertObservationMaintainsPacketSites|InsertObservationResendWritesNothing|PacketSitesRaceMaxMerge|InsertObservationSiteFallback|PacketSitesInitAndBackfill|FillPacketSitesSliceIdempotent|FillPacketSitesSliceRetries23503|PacketSitesReconcile|PacketSitesAudit|PacketSitesAuditTakesNewestHearing|PacketsBySitesMatchesTruth|PacketsBySitesCursorTies|PacketsBySitesManyBogusSites|PacketsBySitesSaturatedPagesFull|PacketsBySitesMatchesLegacy|PacketSitesTrustLifecycle|PacketsBySitesBounded|PacketsBySitesFallbackKeepsPageCursor|SweepPacketSitesFrontsPlan|IndexHealthPacketSitesFreshBaseline|PacketSitesFirstVerification|PacedBackfillSlices|FillSessionsPlanCustom|FillAnalyzeFailureIsIgnored|FillAnalyzeOnlyNeverAnalyzed|HygieneIndexSet|HygieneReadersAfterIndexDrops|KnownRoutesColumns|SequenceUse|SlowPoolIsolation|SlowPoolSize|SequenceUseMixedCaseSchema|ObservationsIndexSet|ReadersAfterIndexDrops)Postgres$' -count=1 -v
 
       - name: Verify the channel message site rule against PostgreSQL 16
         env:

db/observer_comparison_plan_integration_test.go

@@ -0,0 +1,226 @@
+// Copyright 2026 Beacon Contributors
+// SPDX-License-Identifier: AGPL-3.0-or-later
+
+package db
+
+import (
+	"encoding/json"
+	"fmt"
+	"strings"
+	"testing"
+	"time"
+
+	sqlc "github.com/MeshCore-Beacon/beacon-server/db/sqlc"
+	"github.com/google/uuid"
+	"github.com/jackc/pgx/v5/pgtype"
+)
+
+// legacyGetObserverComparison is the comparison before each observer's receptions were read on their own,
+// verbatim from the generated db/sqlc/observer_comparison.sql.go. The statement changed in place.
+const legacyGetObserverComparison = `-- name: GetObserverComparison :one
+WITH heard AS (
+    SELECT po.packet_hash,
+           bool_or(po.observer_id = $1::uuid) AS seen_a,
+           bool_or(po.observer_id = $2::uuid) AS seen_b
+    FROM packet_observations po
+    JOIN packets p ON p.packet_hash = po.packet_hash
+    WHERE po.observer_id IN ($1::uuid, $2::uuid)
+      AND po.heard_at >= $3::timestamptz
+      AND po.heard_at < $4::timestamptz
+      AND p.route_type IN (0, 1)
+      AND (COALESCE(cardinality($5::text[]), 0) = 0 OR po.iata = ANY($5::text[]))
+    GROUP BY po.packet_hash
+)
+SELECT count(*) AS total_packets,
+       count(*) FILTER (WHERE seen_a AND NOT seen_b) AS only_a,
+       count(*) FILTER (WHERE seen_b AND NOT seen_a) AS only_b,
+       count(*) FILTER (WHERE seen_a AND seen_b) AS both,
+       EXISTS (SELECT 1 FROM observers WHERE id = $1::uuid) AS known_a,
+       EXISTS (SELECT 1 FROM observers WHERE id = $2::uuid) AS known_b
+FROM heard
+`
+
+var comparisonBase = time.Date(2026, 10, 1, 0, 0, 0, 0, time.UTC)
+
+// comparisonFixture: 2,000 packets of every route type, one every five minutes from comparisonBase (about seven
+// days), heard by eight observers with skewed rates (90 % down to 2 %) at a per-observer offset of whole seconds, so
+// many receptions share an instant. 100,000 older packets that nobody heard make reading all of packets cost what it
+// does at size. Observer 1 hears at YVR and sometimes YYJ, observer 4 alternates YVR and YYZ,
+// observers 5-8 hear at YOW. Observer 9 is known and never hears anything.
+var comparisonFixture = []string{`
+INSERT INTO observers (id, public_key)
+SELECT ('00000000-0000-0000-0000-' || lpad(i::text, 12, '0'))::uuid, decode(lpad(to_hex(i), 64, '0'), 'hex')
+FROM generate_series(1, 9) i WHERE $1::timestamptz IS NOT NULL`, `
+INSERT INTO packets (packet_hash, payload_type, payload_version, route_type, raw_payload, raw_header, first_heard_at, last_heard_at)
+SELECT decode(md5('cmp-' || i), 'hex'), i % 12, 0, (ARRAY[0, 1, 1, 0, 2, 1, 3, 0])[i % 8 + 1], '\x00', '\x00',
+       $1::timestamptz + i * interval '5 minutes', $1::timestamptz + i * interval '5 minutes'
+FROM generate_series(1, 2000) i`, `
+INSERT INTO packets (packet_hash, payload_type, payload_version, route_type, raw_payload, raw_header, first_heard_at, last_heard_at)
+SELECT decode(md5('cmp-unheard-' || i), 'hex'), i % 12, 0, i % 4, '\x00', '\x00',
+       $1::timestamptz - interval '30 days' + i * interval '1 second', $1::timestamptz - interval '30 days' + i * interval '1 second'
+FROM generate_series(1, 100000) i`, `
+INSERT INTO packet_observations (packet_hash, observer_id, iata, heard_at, path_length_byte, hash_size, hop_count, source_broker)
+SELECT decode(md5('cmp-' || i), 'hex'),
+       ('00000000-0000-0000-0000-' || lpad(o::text, 12, '0'))::uuid,
+       CASE WHEN o = 1 AND i % 7 = 0 THEN 'YYJ' WHEN o IN (1, 2) THEN 'YVR' WHEN o = 3 THEN 'YYJ'
+            WHEN o = 4 THEN (ARRAY['YVR', 'YYZ'])[i % 2 + 1] ELSE 'YOW' END,
+       $1::timestamptz + i * interval '5 minutes' + (o % 3) * interval '1 second',
+       0, 1, 0, 'a'
+FROM generate_series(1, 2000) i
+CROSS JOIN generate_series(1, 8) o
+WHERE ('x' || substr(md5(i || '-' || o), 1, 8))::bit(32)::bigint % 100 < (ARRAY[90, 70, 40, 10, 2, 2, 2, 2])[o]`}
+
+// The comparison reads each observer's receptions from idx_observations_observer and returns exactly what the
+// statement it replaced returned, in custom and generic plans, also with sequential scans off.
+func TestObserverComparisonMatchesTodayPostgres(t *testing.T) {
+	ctx, tx := retentionTx(t)
+	applyBaseline(t, ctx, tx)
+	// Migration 046 drops it; nothing may lean on it.
+	mustExec(t, ctx, tx, "DROP INDEX idx_observations_observer_id")
+	for _, sql := range comparisonFixture {
+		if _, err := tx.Exec(ctx, sql, comparisonBase); err != nil {
+			t.Fatal(err)
+		}
+	}
+	mustExec(t, ctx, tx, "ANALYZE observers; ANALYZE packets; ANALYZE packet_observations")
+	mustExec(t, ctx, tx, "SET LOCAL jit = off")
+
+	obs := func(i int) uuid.UUID { return uuid.MustParse(fmt.Sprintf("00000000-0000-0000-0000-%012d", i)) }
+	unknown := uuid.MustParse("00000000-0000-0000-0000-0000000000ff")
+	pairs := [][2]uuid.UUID{
+		{obs(1), obs(2)}, {obs(2), obs(1)}, {obs(1), obs(4)}, {obs(3), obs(4)}, {obs(4), obs(5)},
+		{obs(5), obs(6)}, {obs(1), obs(9)}, {obs(9), obs(5)}, {obs(1), obs(1)}, {obs(2), unknown},
+		{unknown, obs(1)},
+	}
+	// One reception instant, to put window edges exactly on receptions.
+	edge := comparisonBase.Add(500*5*time.Minute + time.Second)
+	ts := func(at time.Time) pgtype.Timestamptz { return pgtype.Timestamptz{Time: at, Valid: true} }
+	windows := [][2]time.Time{
+		{comparisonBase.Add(-time.Hour), comparisonBase.Add(8 * 24 * time.Hour)},
+		{comparisonBase.Add(24 * time.Hour), comparisonBase.Add(48 * time.Hour)},
+		{edge, edge.Add(time.Hour)},
+		{edge.Add(-time.Hour), edge},
+		{edge, edge.Add(time.Nanosecond * 1000)},
+		{comparisonBase.Add(30 * 24 * time.Hour), comparisonBase.Add(31 * 24 * time.Hour)},
+	}
+	iataSets := [][]string{nil, {}, {"YVR"}, {"YYJ", "YVR", "YYJ"}, {"YOW"}, {"ZZZ"}, {""}, {"yvr"}}
+
+	q := sqlc.New(tx)
+	settings := []string{"RESET enable_seqscan", "SET LOCAL enable_seqscan = off"}
+	compared, nonEmpty := 0, 0
+	for _, setting := range settings {
+		for _, mode := range []string{"force_custom_plan", "force_generic_plan"} {
+			mustExec(t, ctx, tx, setting)
+			mustExec(t, ctx, tx, "SET LOCAL plan_cache_mode = "+mode)
+			// Cached plans keep the settings they were made under.
+			if err := tx.Conn().DeallocateAll(ctx); err != nil {
+				t.Fatal(err)
+			}
+			for _, pair := range pairs {
+				for _, w := range windows {
+					for _, iatas := range iataSets {
+						p := sqlc.GetObserverComparisonParams{
+							ObserverA: pair[0], ObserverB: pair[1], Since: ts(w[0]), Until: ts(w[1]), Iatas: iatas,
+						}
+						got, err := q.GetObserverComparison(ctx, p)
+						if err != nil {
+							t.Fatal(err)
+						}
+						want := legacyRows[sqlc.GetObserverComparisonRow](t, ctx, tx, legacyGetObserverComparison,
+							p.ObserverA, p.ObserverB, p.Since, p.Until, p.Iatas)
+						if len(want) != 1 || got != want[0] {
+							t.Fatalf("%s, %s: GetObserverComparison(%+v) = %+v, today %+v", setting, mode, p, got, want)
+						}
+						compared++
+						if got.TotalPackets > 0 {
+							nonEmpty++
+						}
+					}
+				}
+			}
+		}
+	}
+	t.Logf("%d parameter sets compared, %d non-empty", compared, nonEmpty)
+	if nonEmpty < compared/4 {
+		t.Fatalf("only %d of %d parameter sets counted packets; the grid is too sparse to compare", nonEmpty, compared)
+	}
+
+	// Plans: every observation scan is one observer's range of idx_observations_observer, never a walk of the window,
+	// and packets is only probed through packets_pkey once per heard packet, never scanned whole.
+	capture := &statementCapture{Tx: tx, name: "GetObserverComparison"}
+	if _, err := sqlc.New(capture).GetObserverComparison(ctx, sqlc.GetObserverComparisonParams{ObserverA: obs(1), ObserverB: obs(2)}); err != nil {
+		t.Fatal(err)
+	}
+	if capture.statement == "" {
+		t.Fatal("GetObserverComparison statement not captured")
+	}
+	mustExec(t, ctx, tx, "SET LOCAL enable_seqscan = off")
+	mustExec(t, ctx, tx, "PREPARE observer_comparison AS "+capture.statement)
+	for _, mode := range []string{"force_custom_plan", "force_generic_plan"} {
+		mustExec(t, ctx, tx, "SET LOCAL plan_cache_mode = "+mode)
+		for _, pair := range [][2]int{{1, 2}, {5, 6}, {1, 9}} {
+			execute := fmt.Sprintf("EXPLAIN (FORMAT JSON) EXECUTE observer_comparison('%s', '%s', '%s', '%s', NULL)",
+				obs(pair[0]), obs(pair[1]), comparisonBase.Format(time.RFC3339), comparisonBase.Add(7*24*time.Hour).Format(time.RFC3339))
+			var raw []byte
+			if err := tx.QueryRow(ctx, execute).Scan(&raw); err != nil {
+				t.Fatal(err)
+			}
+			var plans []struct{ Plan map[string]any }
+			if err := json.Unmarshal(raw, &plans); err != nil || len(plans) != 1 {
+				t.Fatalf("EXPLAIN JSON did not parse (%v): %s", err, raw)
+			}
+			var scans, packetScans []map[string]any
+			var packetParents []any
+			var walk func(n, parent map[string]any)
+			walk = func(n, parent map[string]any) {
+				index, _ := n["Index Name"].(string)
+				bitmap := n["Node Type"] == "Bitmap Index Scan"
+				switch {
+				case n["Relation Name"] == "packet_observations" ||
+					(bitmap && (strings.HasPrefix(index, "idx_observations") || strings.HasPrefix(index, "packet_observations"))):
+					scans = append(scans, n)
+				case n["Relation Name"] == "packets" || (bitmap && (strings.HasPrefix(index, "idx_packets") || index == "packets_pkey")):
+					packetScans = append(packetScans, n)
+					packetParents = append(packetParents, parent["Node Type"])
+				}
+				if children, ok := n["Plans"].([]any); ok {
+					for _, c := range children {
+						walk(c.(map[string]any), n)
+					}
+				}
+			}
+			walk(plans[0].Plan, nil)
+			indexScans := 0
+			for _, s := range scans {
+				switch s["Node Type"] {
+				case "Bitmap Heap Scan":
+					continue // its Bitmap Index Scan child names the index
+				case "Index Scan", "Index Only Scan", "Bitmap Index Scan":
+					cond, _ := s["Index Cond"].(string)
+					if s["Index Name"] != "idx_observations_observer" || !strings.Contains(cond, "observer_id =") {
+						t.Errorf("%s %v: observations read through %v (%s), not one observer's idx_observations_observer range: %s",
+							mode, pair, s["Index Name"], cond, raw)
+					}
+					indexScans++
+				default:
+					t.Errorf("%s %v: observations read by %v: %s", mode, pair, s["Node Type"], raw)
+				}
+			}
+			if indexScans != 2 {
+				t.Errorf("%s %v: %d observation index scans, want one per observer: %s", mode, pair, indexScans, raw)
+			}
+			if len(packetScans) != 1 {
+				t.Errorf("%s %v: %d scans of packets, want one probe: %s", mode, pair, len(packetScans), raw)
+			}
+			for i, s := range packetScans {
+				cond, _ := s["Index Cond"].(string)
+				if s["Node Type"] != "Index Scan" || s["Index Name"] != "packets_pkey" ||
+					!strings.Contains(cond, "packet_hash = COALESCE") || packetParents[i] != "Nested Loop" {
+					t.Errorf("%s %v: packets read by %v on %v (%s) under %v, not a packets_pkey probe per heard packet: %s",
+						mode, pair, s["Node Type"], s["Index Name"], cond, packetParents[i], raw)
+				}
+			}
+		}
+	}
+	mustExec(t, ctx, tx, "DEALLOCATE observer_comparison")
+}

db/planguard/baseline/fixture/obscmp.busy_24h_all.custom.json

@@ -0,0 +1,13 @@
+[
+ "Aggregate partial=Simple",
+ "  Index Only Scan on observers using observers_pkey scan=Forward [InitPlan]",
+ "  Index Only Scan on observers observers_1 using observers_pkey scan=Forward [InitPlan]",
+ "  Result [InitPlan]",
+ "  Result [InitPlan]",
+ "  Nested Loop",
+ "    Hash Join (Full)",
+ "      Index Scan on packet_observations po using idx_observations_observer scan=Forward",
+ "      Hash",
+ "        Index Scan on packet_observations po_1 using idx_observations_observer scan=Forward",
+ "    Index Scan on packets p using packets_pkey scan=Forward"
+]

db/planguard/baseline/fixture/obscmp.busy_24h_site.custom.json

@@ -0,0 +1,13 @@
+[
+ "Aggregate partial=Simple",
+ "  Index Only Scan on observers using observers_pkey scan=Forward [InitPlan]",
+ "  Index Only Scan on observers observers_1 using observers_pkey scan=Forward [InitPlan]",
+ "  Result [InitPlan]",
+ "  Result [InitPlan]",
+ "  Nested Loop",
+ "    Hash Join (Full)",
+ "      Index Scan on packet_observations po using idx_observations_observer scan=Forward",
+ "      Hash",
+ "        Index Scan on packet_observations po_1 using idx_observations_observer scan=Forward",
+ "    Index Scan on packets p using packets_pkey scan=Forward"
+]

db/planguard/baseline/fixture/obscmp.busy_7d_all.custom.json

@@ -0,0 +1,13 @@
+[
+ "Aggregate partial=Simple",
+ "  Index Only Scan on observers using observers_pkey scan=Forward [InitPlan]",
+ "  Index Only Scan on observers observers_1 using observers_pkey scan=Forward [InitPlan]",
+ "  Result [InitPlan]",
+ "  Result [InitPlan]",
+ "  Nested Loop",
+ "    Hash Join (Full)",
+ "      Index Scan on packet_observations po using idx_observations_observer scan=Forward",
+ "      Hash",
+ "        Index Scan on packet_observations po_1 using idx_observations_observer scan=Forward",
+ "    Index Scan on packets p using packets_pkey scan=Forward"
+]

db/planguard/baseline/fixture/obscmp.busy_7d_empty_region.custom.json

@@ -0,0 +1,13 @@
+[
+ "Aggregate partial=Simple",
+ "  Index Only Scan on observers using observers_pkey scan=Forward [InitPlan]",
+ "  Index Only Scan on observers observers_1 using observers_pkey scan=Forward [InitPlan]",
+ "  Result [InitPlan]",
+ "  Result [InitPlan]",
+ "  Nested Loop",
+ "    Hash Join (Full)",
+ "      Index Scan on packet_observations po using idx_observations_observer scan=Forward",
+ "      Hash",
+ "        Index Scan on packet_observations po_1 using idx_observations_observer scan=Forward",
+ "    Index Scan on packets p using packets_pkey scan=Forward"
+]

db/planguard/baseline/fixture/obscmp.busy_7d_region.custom.json

@@ -0,0 +1,13 @@
+[
+ "Aggregate partial=Simple",
+ "  Index Only Scan on observers using observers_pkey scan=Forward [InitPlan]",
+ "  Index Only Scan on observers observers_1 using observers_pkey scan=Forward [InitPlan]",
+ "  Result [InitPlan]",
+ "  Result [InitPlan]",
+ "  Nested Loop",
+ "    Hash Join (Full)",
+ "      Index Scan on packet_observations po using idx_observations_observer scan=Forward",
+ "      Hash",
+ "        Index Scan on packet_observations po_1 using idx_observations_observer scan=Forward",
+ "    Index Scan on packets p using packets_pkey scan=Forward"
+]

db/planguard/baseline/fixture/obscmp.mixed_7d_all.custom.json

@@ -0,0 +1,13 @@
+[
+ "Aggregate partial=Simple",
+ "  Index Only Scan on observers using observers_pkey scan=Forward [InitPlan]",
+ "  Index Only Scan on observers observers_1 using observers_pkey scan=Forward [InitPlan]",
+ "  Result [InitPlan]",
+ "  Result [InitPlan]",
+ "  Nested Loop",
+ "    Hash Join (Full)",
+ "      Index Scan on packet_observations po using idx_observations_observer scan=Forward",
+ "      Hash",
+ "        Index Scan on packet_observations po_1 using idx_observations_observer scan=Forward",
+ "    Index Scan on packets p using packets_pkey scan=Forward"
+]

db/planguard/baseline/fixture/obscmp.quiet_7d_all.custom.json

@@ -0,0 +1,13 @@
+[
+ "Aggregate partial=Simple",
+ "  Index Only Scan on observers using observers_pkey scan=Forward [InitPlan]",
+ "  Index Only Scan on observers observers_1 using observers_pkey scan=Forward [InitPlan]",
+ "  Result [InitPlan]",
+ "  Result [InitPlan]",
+ "  Nested Loop",
+ "    Hash Join (Full)",
+ "      Index Scan on packet_observations po using idx_observations_observer scan=Forward",
+ "      Hash",
+ "        Index Scan on packet_observations po_1 using idx_observations_observer scan=Forward",
+ "    Index Scan on packets p using packets_pkey scan=Forward"
+]

db/planguard/baseline/fixture/obscmp.unknown_7d.custom.json

@@ -0,0 +1,13 @@
+[
+ "Aggregate partial=Simple",
+ "  Index Only Scan on observers using observers_pkey scan=Forward [InitPlan]",
+ "  Index Only Scan on observers observers_1 using observers_pkey scan=Forward [InitPlan]",
+ "  Result [InitPlan]",
+ "  Result [InitPlan]",
+ "  Nested Loop",
+ "    Hash Join (Full)",
+ "      Index Scan on packet_observations po using idx_observations_observer scan=Forward",
+ "      Hash",
+ "        Index Scan on packet_observations po_1 using idx_observations_observer scan=Forward",
+ "    Index Scan on packets p using packets_pkey scan=Forward"
+]

db/planguard/baseline/local/obscmp.busy_24h_all.custom.json

@@ -0,0 +1,13 @@
+[
+ "Aggregate partial=Simple",
+ "  Index Only Scan on observers using observers_pkey scan=Forward [InitPlan]",
+ "  Index Only Scan on observers observers_1 using observers_pkey scan=Forward [InitPlan]",
+ "  Result [InitPlan]",
+ "  Result [InitPlan]",
+ "  Nested Loop",
+ "    Hash Join (Full)",
+ "      Index Scan on packet_observations po using idx_observations_observer scan=Forward",
+ "      Hash",
+ "        Index Scan on packet_observations po_1 using idx_observations_observer scan=Forward",
+ "    Index Scan on packets p using packets_pkey scan=Forward"
+]

db/planguard/baseline/local/obscmp.busy_24h_site.custom.json

@@ -0,0 +1,13 @@
+[
+ "Aggregate partial=Simple",
+ "  Index Only Scan on observers using observers_pkey scan=Forward [InitPlan]",
+ "  Index Only Scan on observers observers_1 using observers_pkey scan=Forward [InitPlan]",
+ "  Result [InitPlan]",
+ "  Result [InitPlan]",
+ "  Nested Loop",
+ "    Hash Join (Full)",
+ "      Index Scan on packet_observations po using idx_observations_observer scan=Forward",
+ "      Hash",
+ "        Index Scan on packet_observations po_1 using idx_observations_observer scan=Forward",
+ "    Index Scan on packets p using packets_pkey scan=Forward"
+]

db/planguard/baseline/local/obscmp.busy_7d_all.custom.json

@@ -0,0 +1,13 @@
+[
+ "Aggregate partial=Simple",
+ "  Index Only Scan on observers using observers_pkey scan=Forward [InitPlan]",
+ "  Index Only Scan on observers observers_1 using observers_pkey scan=Forward [InitPlan]",
+ "  Result [InitPlan]",
+ "  Result [InitPlan]",
+ "  Nested Loop",
+ "    Hash Join (Full)",
+ "      Index Scan on packet_observations po using idx_observations_observer scan=Forward",
+ "      Hash",
+ "        Index Scan on packet_observations po_1 using idx_observations_observer scan=Forward",
+ "    Index Scan on packets p using packets_pkey scan=Forward"
+]

db/planguard/baseline/local/obscmp.busy_7d_empty_region.custom.json

@@ -0,0 +1,13 @@
+[
+ "Aggregate partial=Simple",
+ "  Index Only Scan on observers using observers_pkey scan=Forward [InitPlan]",
+ "  Index Only Scan on observers observers_1 using observers_pkey scan=Forward [InitPlan]",
+ "  Result [InitPlan]",
+ "  Result [InitPlan]",
+ "  Nested Loop",
+ "    Hash Join (Full)",
+ "      Index Scan on packet_observations po using idx_observations_observer scan=Forward",
+ "      Hash",
+ "        Index Scan on packet_observations po_1 using idx_observations_observer scan=Forward",
+ "    Index Scan on packets p using packets_pkey scan=Forward"
+]

db/planguard/baseline/local/obscmp.busy_7d_region.custom.json

@@ -0,0 +1,13 @@
+[
+ "Aggregate partial=Simple",
+ "  Index Only Scan on observers using observers_pkey scan=Forward [InitPlan]",
+ "  Index Only Scan on observers observers_1 using observers_pkey scan=Forward [InitPlan]",
+ "  Result [InitPlan]",
+ "  Result [InitPlan]",
+ "  Nested Loop",
+ "    Hash Join (Full)",
+ "      Index Scan on packet_observations po using idx_observations_observer scan=Forward",
+ "      Hash",
+ "        Index Scan on packet_observations po_1 using idx_observations_observer scan=Forward",
+ "    Index Scan on packets p using packets_pkey scan=Forward"
+]

db/planguard/baseline/local/obscmp.mixed_7d_all.custom.json

@@ -0,0 +1,13 @@
+[
+ "Aggregate partial=Simple",
+ "  Index Only Scan on observers using observers_pkey scan=Forward [InitPlan]",
+ "  Index Only Scan on observers observers_1 using observers_pkey scan=Forward [InitPlan]",
+ "  Result [InitPlan]",
+ "  Result [InitPlan]",
+ "  Nested Loop",
+ "    Hash Join (Full)",
+ "      Index Scan on packet_observations po using idx_observations_observer scan=Forward",
+ "      Hash",
+ "        Index Scan on packet_observations po_1 using idx_observations_observer scan=Forward",
+ "    Index Scan on packets p using packets_pkey scan=Forward"
+]

db/planguard/baseline/local/obscmp.quiet_7d_all.custom.json

@@ -0,0 +1,13 @@
+[
+ "Aggregate partial=Simple",
+ "  Index Only Scan on observers using observers_pkey scan=Forward [InitPlan]",
+ "  Index Only Scan on observers observers_1 using observers_pkey scan=Forward [InitPlan]",
+ "  Result [InitPlan]",
+ "  Result [InitPlan]",
+ "  Nested Loop",
+ "    Hash Join (Full)",
+ "      Index Scan on packet_observations po using idx_observations_observer scan=Forward",
+ "      Hash",
+ "        Index Scan on packet_observations po_1 using idx_observations_observer scan=Forward",
+ "    Index Scan on packets p using packets_pkey scan=Forward"
+]

db/planguard/baseline/local/obscmp.unknown_7d.custom.json

@@ -0,0 +1,13 @@
+[
+ "Aggregate partial=Simple",
+ "  Index Only Scan on observers using observers_pkey scan=Forward [InitPlan]",
+ "  Index Only Scan on observers observers_1 using observers_pkey scan=Forward [InitPlan]",
+ "  Result [InitPlan]",
+ "  Result [InitPlan]",
+ "  Nested Loop",
+ "    Hash Join (Full)",
+ "      Index Scan on packet_observations po using idx_observations_observer scan=Forward",
+ "      Hash",
+ "        Index Scan on packet_observations po_1 using idx_observations_observer scan=Forward",
+ "    Index Scan on packets p using packets_pkey scan=Forward"
+]

db/planguard/corpus.yaml

@@ -108,6 +108,7 @@ statements:
   GetStatsTopAdvertisers: {sqlc: GetStatsTopAdvertisers, plan: custom}
   GetPathStats: {sqlc: GetPathStats, plan: custom}
   GetStatsSeries: {sqlc: GetStatsSeries, group: stats_adverts, plan: custom}
+  GetObserverComparison: {sqlc: GetObserverComparison, plan: custom}
   ListTraceTags: {sqlc: ListTraceTags, group: stats_adverts, plan: custom}
   # One observer's adverts and telemetry pages.
   ListObserverAdverts: {sqlc: ListObserverAdverts, group: stats_adverts, plan: cached, admission: observer_page}
@@ -244,6 +245,19 @@ resolvers:
   ps_sparse_site: "SELECT iata FROM (SELECT po.iata, count(*) n FROM packet_observations po WHERE po.id > (SELECT max(id) - 200000 FROM packet_observations) GROUP BY po.iata ORDER BY n DESC, po.iata) r ORDER BY n, iata LIMIT 1"
   ps_stale_site: "SELECT iata FROM packet_sites WHERE iata = 'NEW' LIMIT 1"
   ps_zero_scope: "SELECT ARRAY[ts.name] FROM transport_scopes ts WHERE NOT EXISTS (SELECT 1 FROM packet_sites ps WHERE ps.iata = (SELECT iata FROM (SELECT po.iata, count(*) n FROM packet_observations po WHERE po.id > (SELECT max(id) - 200000 FROM packet_observations) GROUP BY po.iata ORDER BY n DESC, po.iata LIMIT 1) r) AND ps.scope_id = ts.id) ORDER BY ts.name LIMIT 1"
+  # Observer comparison roles, picked among the observers in the newest 200,000 observation ids (a pkey range): the
+  # two busiest, the busiest one's latest site, and the two that heard least (but something) in the 7 d window, each
+  # counted on idx_observations_observer up to 1,000. cmp_unknown is a fixed non-zero id that matches no observer.
+  # Windows end at the newest reception (a max on idx_observations_heard).
+  cmp_busy1: "SELECT observer_id FROM packet_observations WHERE id > (SELECT max(id) FROM packet_observations) - 200000 GROUP BY 1 ORDER BY count(*) DESC, observer_id LIMIT 1"
+  cmp_busy2: "SELECT observer_id FROM packet_observations WHERE id > (SELECT max(id) FROM packet_observations) - 200000 GROUP BY 1 ORDER BY count(*) DESC, observer_id OFFSET 1 LIMIT 1"
+  cmp_quiet1_7d: "WITH t AS (SELECT max(heard_at) hi FROM packet_observations), c AS (SELECT DISTINCT observer_id FROM packet_observations WHERE id > (SELECT max(id) FROM packet_observations) - 200000) SELECT c.observer_id FROM c CROSS JOIN t CROSS JOIN LATERAL (SELECT count(*) n FROM (SELECT 1 FROM packet_observations x WHERE x.observer_id = c.observer_id AND x.heard_at >= t.hi - interval '7 days' AND x.heard_at < t.hi LIMIT 1000) s) k WHERE k.n > 0 ORDER BY k.n, c.observer_id LIMIT 1"
+  cmp_quiet2_7d: "WITH t AS (SELECT max(heard_at) hi FROM packet_observations), c AS (SELECT DISTINCT observer_id FROM packet_observations WHERE id > (SELECT max(id) FROM packet_observations) - 200000) SELECT c.observer_id FROM c CROSS JOIN t CROSS JOIN LATERAL (SELECT count(*) n FROM (SELECT 1 FROM packet_observations x WHERE x.observer_id = c.observer_id AND x.heard_at >= t.hi - interval '7 days' AND x.heard_at < t.hi LIMIT 1000) s) k WHERE k.n > 0 ORDER BY k.n, c.observer_id OFFSET 1 LIMIT 1"
+  cmp_busy1_site: "SELECT iata FROM packet_observations WHERE id > (SELECT max(id) FROM packet_observations) - 200000 AND observer_id = (SELECT observer_id FROM packet_observations WHERE id > (SELECT max(id) FROM packet_observations) - 200000 GROUP BY 1 ORDER BY count(*) DESC, observer_id LIMIT 1) ORDER BY id DESC LIMIT 1"
+  cmp_unknown: "SELECT md5('observer comparison: no such observer')::uuid WHERE NOT EXISTS (SELECT 1 FROM observers WHERE id = md5('observer comparison: no such observer')::uuid)"
+  cmp_until: "SELECT max(heard_at) FROM packet_observations"
+  cmp_since_1d: "SELECT max(heard_at) - interval '1 day' FROM packet_observations"
+  cmp_since_7d: "SELECT max(heard_at) - interval '7 days' FROM packet_observations"
   # The adverts and telemetry roles are picked among the newest 5,000,000 observation ids (200,000 telemetry ids, about
   # 10K rows), a pkey range: advert counts are aggregate filters, never a WHERE on payload_type, so the partial adverts
   # index cannot replace the window. They stay far below the prod tier's 20 s cap; adv_zero and adv_quiet then probe
@@ -2404,6 +2418,80 @@ entries:
       budget: {buffers: 520, planning_ms: 10}
     note: "measured: 3 / 3 B (custom / generic); budget: gate <= 10 B/row + 10"
 
+  # Observer comparison: each observer's receptions in the window from idx_observations_observer, then one packets_pkey
+  # probe per heard packet; never a walk of the window through idx_observations_heard, never a scan of all packets.
+  - id: obscmp.busy_24h_all
+    stmt: GetObserverComparison
+    shape: "the two busiest observers, 24 h, every site"
+    params: [{r: cmp_busy1}, {r: cmp_busy2}, {r: cmp_since_1d}, {r: cmp_until}, null]
+    expect:
+      driver: [{relation: packet_observations, index: idx_observations_observer, nodes: ["Bitmap Heap Scan", "Index Scan"], index_cond: ["observer_id ="]},
+               {relation: packets, index: packets_pkey, nodes: ["Index Scan"], index_cond: ["packet_hash = COALESCE"]}]
+      budget: {buffers: 196341, planning_ms: 10}
+    note: "measured: 130894 B on the local copy (t_end, custom; the statement it replaced: 130894 B); budget: 1.5x measured"
+  - id: obscmp.busy_7d_all
+    stmt: GetObserverComparison
+    shape: "the two busiest observers, 7 d, every site"
+    params: [{r: cmp_busy1}, {r: cmp_busy2}, {r: cmp_since_7d}, {r: cmp_until}, null]
+    expect:
+      driver: [{relation: packet_observations, index: idx_observations_observer, nodes: ["Bitmap Heap Scan", "Index Scan"], index_cond: ["observer_id ="]},
+               {relation: packets, index: packets_pkey, nodes: ["Index Scan"], index_cond: ["packet_hash = COALESCE"]}]
+      budget: {buffers: 558295, planning_ms: 10}
+    note: "measured: 372197 B on the local copy (t_end, custom; the statement it replaced: 370407 B); budget: 1.5x measured"
+  - id: obscmp.busy_24h_site
+    stmt: GetObserverComparison
+    shape: "the two busiest observers, 24 h, the busiest one's site"
+    params: [{r: cmp_busy1}, {r: cmp_busy2}, {r: cmp_since_1d}, {r: cmp_until}, [{r: cmp_busy1_site}]]
+    expect:
+      driver: [{relation: packet_observations, index: idx_observations_observer, nodes: ["Bitmap Heap Scan", "Index Scan"], index_cond: ["observer_id ="]},
+               {relation: packets, index: packets_pkey, nodes: ["Index Scan"], index_cond: ["packet_hash = COALESCE"]}]
+      budget: {buffers: 91743, planning_ms: 10}
+    note: "measured: 61162 B on the local copy (t_end, custom; the statement it replaced: 61162 B); budget: 1.5x measured"
+  - id: obscmp.busy_7d_region
+    stmt: GetObserverComparison
+    shape: "the two busiest observers, 7 d, the seven busiest sites"
+    params: [{r: cmp_busy1}, {r: cmp_busy2}, {r: cmp_since_7d}, {r: cmp_until}, {r: ps_busy_sites7}]
+    expect:
+      driver: [{relation: packet_observations, index: idx_observations_observer, nodes: ["Bitmap Heap Scan", "Index Scan"], index_cond: ["observer_id ="]},
+               {relation: packets, index: packets_pkey, nodes: ["Index Scan"], index_cond: ["packet_hash = COALESCE"]}]
+      budget: {buffers: 258049, planning_ms: 10}
+    note: "measured: 172033 B on the local copy (t_end, custom; the statement it replaced: 170243 B); budget: 1.5x measured"
+  - id: obscmp.busy_7d_empty_region
+    stmt: GetObserverComparison
+    shape: "the two busiest observers, 7 d, a region with no sites"
+    params: [{r: cmp_busy1}, {r: cmp_busy2}, {r: cmp_since_7d}, {r: cmp_until}, [""]]
+    expect:
+      driver: [{relation: packet_observations, index: idx_observations_observer, nodes: ["Bitmap Heap Scan", "Index Scan"], index_cond: ["observer_id ="]},
+               {relation: packets, index: packets_pkey, nodes: ["Index Scan"], index_cond: ["packet_hash = COALESCE"]}]
+      budget: {buffers: 101311, planning_ms: 10}
+    note: "measured: 67541 B on the local copy (t_end, custom; the statement it replaced: 65751 B); budget: 1.5x measured"
+  - id: obscmp.quiet_7d_all
+    stmt: GetObserverComparison
+    shape: "the two quietest observers, 7 d, every site"
+    params: [{r: cmp_quiet1_7d}, {r: cmp_quiet2_7d}, {r: cmp_since_7d}, {r: cmp_until}, null]
+    expect:
+      driver: [{relation: packet_observations, index: idx_observations_observer, nodes: ["Bitmap Heap Scan", "Index Scan"], index_cond: ["observer_id ="]},
+               {relation: packets, index: packets_pkey, nodes: ["Index Scan"], index_cond: ["packet_hash = COALESCE"]}]
+      budget: {buffers: 261, planning_ms: 10}
+    note: "measured: 174 B on the local copy (t_end, custom; the statement it replaced: 174 B); budget: 1.5x measured"
+  - id: obscmp.mixed_7d_all
+    stmt: GetObserverComparison
+    shape: "the busiest and the quietest observer, 7 d, every site"
+    params: [{r: cmp_busy1}, {r: cmp_quiet1_7d}, {r: cmp_since_7d}, {r: cmp_until}, null]
+    expect:
+      driver: [{relation: packet_observations, index: idx_observations_observer, nodes: ["Bitmap Heap Scan", "Index Scan"], index_cond: ["observer_id ="]},
+               {relation: packets, index: packets_pkey, nodes: ["Index Scan"], index_cond: ["packet_hash = COALESCE"]}]
+      budget: {buffers: 195237, planning_ms: 10}
+    note: "measured: 130158 B on the local copy (t_end, custom; the statement it replaced: 130158 B); budget: 1.5x measured"
+  - id: obscmp.unknown_7d
+    stmt: GetObserverComparison
+    shape: "the busiest observer and an unknown id, 7 d"
+    params: [{r: cmp_busy1}, {r: cmp_unknown}, {r: cmp_since_7d}, {r: cmp_until}, null]
+    expect:
+      driver: [{relation: packet_observations, index: idx_observations_observer, nodes: ["Bitmap Heap Scan", "Index Scan"], index_cond: ["observer_id ="]},
+               {relation: packets, index: packets_pkey, nodes: ["Index Scan"], index_cond: ["packet_hash = COALESCE"]}]
+      budget: {buffers: 195123, planning_ms: 10}
+    note: "measured: 130082 B on the local copy (t_end, custom; the statement it replaced: 130082 B); budget: 1.5x measured"
   # ---- small tables: nodes, observers, channels, node types, clock drift ----
   - id: sm.nodes.global
     stmt: ListNodes

db/queries/observer_comparison.sql

@@ -1,18 +1,34 @@
 -- name: GetObserverComparison :one
 -- Count packet identities, not reception rows: a broker repeat or another
 -- reception by the same observer must not inflate the comparison.
-WITH heard AS (
-    SELECT po.packet_hash,
-           bool_or(po.observer_id = @observer_a::uuid) AS seen_a,
-           bool_or(po.observer_id = @observer_b::uuid) AS seen_b
+--
+-- Each observer's receptions are read on their own from idx_observations_observer
+-- (observer_id, heard_at): (packet_hash, observer_id) is unique, so each side
+-- holds a packet at most once, and the full join yields each heard packet once.
+-- The route type is checked once per packet after the join.
+-- The observer ids are scalar subqueries so the planner costs every observer
+-- alike: given a busy observer's own row count, a custom plan may instead walk
+-- every reception in the window through idx_observations_heard, which reads
+-- several times the buffers.
+WITH a AS (
+    SELECT po.packet_hash
     FROM packet_observations po
-    JOIN packets p ON p.packet_hash = po.packet_hash
-    WHERE po.observer_id IN (@observer_a::uuid, @observer_b::uuid)
+    WHERE po.observer_id = (SELECT @observer_a::uuid)
       AND po.heard_at >= @since::timestamptz
       AND po.heard_at < @until::timestamptz
-      AND p.route_type IN (0, 1)
       AND (COALESCE(cardinality(@iatas::text[]), 0) = 0 OR po.iata = ANY(@iatas::text[]))
-    GROUP BY po.packet_hash
+), b AS (
+    SELECT po.packet_hash
+    FROM packet_observations po
+    WHERE po.observer_id = (SELECT @observer_b::uuid)
+      AND po.heard_at >= @since::timestamptz
+      AND po.heard_at < @until::timestamptz
+      AND (COALESCE(cardinality(@iatas::text[]), 0) = 0 OR po.iata = ANY(@iatas::text[]))
+), heard AS (
+    SELECT COALESCE(a.packet_hash, b.packet_hash) AS packet_hash,
+           a.packet_hash IS NOT NULL AS seen_a,
+           b.packet_hash IS NOT NULL AS seen_b
+    FROM a FULL JOIN b ON b.packet_hash = a.packet_hash
 )
 SELECT count(*) AS total_packets,
        count(*) FILTER (WHERE seen_a AND NOT seen_b) AS only_a,
@@ -20,4 +36,6 @@ SELECT count(*) AS total_packets,
        count(*) FILTER (WHERE seen_a AND seen_b) AS both,
        EXISTS (SELECT 1 FROM observers WHERE id = @observer_a::uuid) AS known_a,
        EXISTS (SELECT 1 FROM observers WHERE id = @observer_b::uuid) AS known_b
-FROM heard;
+FROM heard
+JOIN packets p ON p.packet_hash = heard.packet_hash
+WHERE p.route_type IN (0, 1);

db/sqlc/observer_comparison.sql.go

@@ -13,18 +13,25 @@ import (
 )
 
 const getObserverComparison = `-- name: GetObserverComparison :one
-WITH heard AS (
-    SELECT po.packet_hash,
-           bool_or(po.observer_id = $1::uuid) AS seen_a,
-           bool_or(po.observer_id = $2::uuid) AS seen_b
+WITH a AS (
+    SELECT po.packet_hash
     FROM packet_observations po
-    JOIN packets p ON p.packet_hash = po.packet_hash
-    WHERE po.observer_id IN ($1::uuid, $2::uuid)
+    WHERE po.observer_id = (SELECT $1::uuid)
       AND po.heard_at >= $3::timestamptz
       AND po.heard_at < $4::timestamptz
-      AND p.route_type IN (0, 1)
       AND (COALESCE(cardinality($5::text[]), 0) = 0 OR po.iata = ANY($5::text[]))
-    GROUP BY po.packet_hash
+), b AS (
+    SELECT po.packet_hash
+    FROM packet_observations po
+    WHERE po.observer_id = (SELECT $2::uuid)
+      AND po.heard_at >= $3::timestamptz
+      AND po.heard_at < $4::timestamptz
+      AND (COALESCE(cardinality($5::text[]), 0) = 0 OR po.iata = ANY($5::text[]))
+), heard AS (
+    SELECT COALESCE(a.packet_hash, b.packet_hash) AS packet_hash,
+           a.packet_hash IS NOT NULL AS seen_a,
+           b.packet_hash IS NOT NULL AS seen_b
+    FROM a FULL JOIN b ON b.packet_hash = a.packet_hash
 )
 SELECT count(*) AS total_packets,
        count(*) FILTER (WHERE seen_a AND NOT seen_b) AS only_a,
@@ -33,6 +40,8 @@ SELECT count(*) AS total_packets,
        EXISTS (SELECT 1 FROM observers WHERE id = $1::uuid) AS known_a,
        EXISTS (SELECT 1 FROM observers WHERE id = $2::uuid) AS known_b
 FROM heard
+JOIN packets p ON p.packet_hash = heard.packet_hash
+WHERE p.route_type IN (0, 1)
 `
 
 type GetObserverComparisonParams struct {
@@ -54,6 +63,15 @@ type GetObserverComparisonRow struct {
 
 // Count packet identities, not reception rows: a broker repeat or another
 // reception by the same observer must not inflate the comparison.
+//
+// Each observer's receptions are read on their own from idx_observations_observer
+// (observer_id, heard_at): (packet_hash, observer_id) is unique, so each side
+// holds a packet at most once, and the full join yields each heard packet once.
+// The route type is checked once per packet after the join.
+// The observer ids are scalar subqueries so the planner costs every observer
+// alike: given a busy observer's own row count, a custom plan may instead walk
+// every reception in the window through idx_observations_heard, which reads
+// several times the buffers.
 func (q *Queries) GetObserverComparison(ctx context.Context, arg GetObserverComparisonParams) (GetObserverComparisonRow, error) {
 	row := q.db.QueryRow(ctx, getObserverComparison,
 		arg.ObserverA,

db/sqlc/querier.go

@@ -148,6 +148,15 @@ type Querier interface {
 	GetObserverByPubkey(ctx context.Context, publicKey []byte) (Observer, error)
 	// Count packet identities, not reception rows: a broker repeat or another
 	// reception by the same observer must not inflate the comparison.
+	//
+	// Each observer's receptions are read on their own from idx_observations_observer
+	// (observer_id, heard_at): (packet_hash, observer_id) is unique, so each side
+	// holds a packet at most once, and the full join yields each heard packet once.
+	// The route type is checked once per packet after the join.
+	// The observer ids are scalar subqueries so the planner costs every observer
+	// alike: given a busy observer's own row count, a custom plan may instead walk
+	// every reception in the window through idx_observations_heard, which reads
+	// several times the buffers.
 	GetObserverComparison(ctx context.Context, arg GetObserverComparisonParams) (GetObserverComparisonRow, error)
 	GetObserverLastIATA(ctx context.Context, id uuid.UUID) (string, error)
 	GetObserverRadio(ctx context.Context, id uuid.UUID) (GetObserverRadioRow, error)

db/statement_capture_test.go

@@ -24,3 +24,10 @@ func (c *statementCapture) Query(ctx context.Context, sql string, args ...any) (
 	}
 	return c.Tx.Query(ctx, sql, args...)
 }
+
+func (c *statementCapture) QueryRow(ctx context.Context, sql string, args ...any) pgx.Row {
+	if strings.Contains(sql, "-- name: "+c.name) {
+		c.statement, c.args = sql, args
+	}
+	return c.Tx.QueryRow(ctx, sql, args...)
+}