amadan

carlos / platform Public

analytics: count pageviews by country with per-day source provenance; attributed sources never fold; one observation per tunnelled response (geoip spec §9, codex R4-1, R7-3)

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
claude 58beffa367c282a1227ca5365ab813d0f1ea1e3c parent de55e92
10 files changed, +345 −7
  • analytics_cmd.go +3 −0
  • apimode.go +2 −0
  • internal/analytics/countries_test.go +99 −0
  • internal/analytics/day.go +63 −4
  • internal/analytics/meter.go +9 −1
  • internal/analytics/read.go +6 −0
  • internal/console/apianalytics.go +43 −1
  • internal/console/apianalytics_geo_test.go +26 −0
  • internal/edge/analytics.go +9 −1
  • internal/edge/tunnel_geo_e2e_test.go +85 −0

CI passed — run details

diff --git a/analytics_cmd.go b/analytics_cmd.go
index fbbfc8161..490277f7c 100644
--- a/analytics_cmd.go
+++ b/analytics_cmd.go
@@ -150,6 +150,9 @@ func printAnalytics(w io.Writer, res apiAnalyticsResponse) {
// Events table
printAnalyticsTable(w, "Events", res.Events)
+
+ // Countries table (geoip spec §9)
+ printAnalyticsTable(w, "Countries", res.Countries)
}
// printAnalyticsTable renders one analytics table (pages, referers, or events)
diff --git a/apimode.go b/apimode.go
index bb873a5d5..fa2f21a66 100644
--- a/apimode.go
+++ b/apimode.go
@@ -3460,6 +3460,8 @@ type apiAnalyticsResponse struct {
Pages map[string]int64 `json:"pages"`
Referers map[string]int64 `json:"referers"`
Events map[string]int64 `json:"events"`
+ Countries map[string]int64 `json:"countries"`
+ GeoSources map[string]int64 `json:"geo_sources"`
PagesApprox bool `json:"pages_approx"`
ReferersApprox bool `json:"referers_approx"`
EventsApprox bool `json:"events_approx"`
diff --git a/internal/analytics/countries_test.go b/internal/analytics/countries_test.go
new file mode 100644
index 000000000..5995fedbd
--- /dev/null
+++ b/internal/analytics/countries_test.go
@@ -0,0 +1,99 @@
+package analytics
+
+import (
+ "context"
+ "encoding/json"
+ "fmt"
+ "testing"
+ "time"
+
+ "github.com/carlosframework/platform/internal/release"
+)
+
+func pv(country, src, state string) Obs {
+ return Obs{Acct: "a", App: "x", Host: "x.test", Method: "GET", Status: 200, ContentType: "text/html",
+ Path: "/", UA: "Mozilla/5.0", ClientIP: "203.0.113.1", When: time.Date(2026, 10, 6, 9, 0, 0, 0, time.UTC),
+ Country: country, GeoSource: src, GeoState: state}
+}
+
+func TestMeterCountsCountriesByState(t *testing.T) {
+ now := time.Date(2026, 10, 6, 9, 0, 0, 0, time.UTC)
+ m := NewMeter(t.TempDir(), "box1", func() time.Time { return now })
+ m.Observe(pv("GB", "DBIP-City-Lite", "hit"))
+ m.Observe(pv("", "DBIP-City-Lite", "hit")) // coordinate-only (codex R3-3)
+ m.Observe(pv("", "DBIP-City-Lite", "miss"))
+ m.Observe(pv("", "", "off"))
+ m.Observe(pv("", "", "")) // an edge from before GeoIP
+ nonPV := pv("FR", "DBIP-City-Lite", "hit")
+ nonPV.Status = 404
+ m.Observe(nonPV)
+ d := m.Drain()[AppKey{Acct: "a", App: "x"}][0]
+ if d.Countries["GB"] != 1 || d.Countries["_unknown"] != 2 || len(d.Countries) != 2 {
+ t.Fatalf("countries = %v", d.Countries)
+ }
+ if d.GeoSources["DBIP-City-Lite"] != 3 || len(d.GeoSources) != 1 {
+ t.Fatalf("sources = %v", d.GeoSources)
+ }
+}
+
+func TestReadRangeMergesCountriesAcrossBoxesDaysAndOldDays(t *testing.T) {
+ ctx := context.Background()
+ s := release.NewDirStore(t.TempDir())
+ put := func(box, date string, d Day) {
+ raw, _ := json.Marshal(d)
+ if _, err := s.Put(ctx, DayKey("a", "x", box, date), raw, ""); err != nil {
+ t.Fatal(err)
+ }
+ }
+ // An old day, from before GeoIP: no geo_sources, empty countries.
+ put("b1", "2026-10-04", Day{V: 1, Date: "2026-10-04", Countries: map[string]int64{}})
+ put("b1", "2026-10-05", Day{V: 1, Date: "2026-10-05", Countries: map[string]int64{"GB": 3}, GeoSources: map[string]int64{"DBIP-City-Lite": 3}})
+ put("b2", "2026-10-05", Day{V: 1, Date: "2026-10-05", Countries: map[string]int64{"GB": 1, "IE": 2}, GeoSources: map[string]int64{"GeoLite2-City": 3}})
+ sum, err := ReadRange(ctx, s, "a", "x", time.Date(2026, 10, 4, 0, 0, 0, 0, time.UTC), time.Date(2026, 10, 5, 0, 0, 0, 0, time.UTC))
+ if err != nil {
+ t.Fatal(err)
+ }
+ if sum.Countries["GB"] != 4 || sum.Countries["IE"] != 2 {
+ t.Fatalf("countries = %v", sum.Countries)
+ }
+ if sum.GeoSources["DBIP-City-Lite"] != 3 || sum.GeoSources["GeoLite2-City"] != 3 {
+ t.Fatalf("both sources must be credited (spec §9): %v", sum.GeoSources)
+ }
+}
+
+func TestDayMergeAddsCountries(t *testing.T) {
+ var d Day
+ d.Merge(Delta{Box: "b", Seq: 1, Countries: map[string]int64{"GB": 1}, GeoSources: map[string]int64{"s": 1}})
+ d.Merge(Delta{Box: "b", Seq: 2, Countries: map[string]int64{"GB": 2, "IE": 1}, GeoSources: map[string]int64{"s": 3}})
+ if d.Countries["GB"] != 3 || d.Countries["IE"] != 1 || d.GeoSources["s"] != 4 {
+ t.Fatalf("day = %v %v", d.Countries, d.GeoSources)
+ }
+}
+
+func TestBoundedAddFoldsIntoOther(t *testing.T) {
+ m := map[string]int64{}
+ for i := 0; i < 12; i++ {
+ boundedAdd(m, fmt.Sprintf("K%02d", i), 1, 8, nil)
+ }
+ if len(m) != 9 || m["_other"] != 4 {
+ t.Fatalf("bounded = %v", m)
+ }
+}
+
+// Codex R7-3: an attributed source is never folded, however many other
+// types crowd the map, across flushes and days.
+func TestAttributedSourcesSurviveOverflow(t *testing.T) {
+ var d Day
+ junk := map[string]int64{}
+ for i := 0; i < 20; i++ {
+ junk[fmt.Sprintf("Custom-%02d", i)] = 1
+ }
+ d.Merge(Delta{Box: "b", Seq: 1, GeoSources: junk})
+ d.Merge(Delta{Box: "b", Seq: 2, GeoSources: map[string]int64{"DBIP-City-Lite": 5, "GeoLite2-Country": 2}})
+ if d.GeoSources["DBIP-City-Lite"] != 5 || d.GeoSources["GeoLite2-Country"] != 2 {
+ t.Fatalf("attributed sources folded: %v", d.GeoSources)
+ }
+ if d.GeoSources["_other"] == 0 {
+ t.Fatalf("unattributed overflow must still fold: %v", d.GeoSources)
+ }
+}
diff --git a/internal/analytics/day.go b/internal/analytics/day.go
index d016afc8f..34728aea8 100644
--- a/internal/analytics/day.go
+++ b/internal/analytics/day.go
@@ -6,6 +6,7 @@ import (
"errors"
"fmt"
"os"
+ "strings"
"github.com/carlosframework/platform/internal/applog"
"github.com/carlosframework/platform/internal/release"
@@ -46,7 +47,10 @@ type Day struct {
Seq map[string]int64 `json:"seq"`
Dropped Dropped `json:"dropped"`
Countries map[string]int64 `json:"countries"`
- Restarted bool `json:"restarted,omitempty"`
+ // GeoSources counts pageviews per database type that answered (geoip
+ // spec §9): the provenance the attribution on the analytics page needs.
+ GeoSources map[string]int64 `json:"geo_sources,omitempty"`
+ Restarted bool `json:"restarted,omitempty"`
}
// Delta is one flush batch from one box for one app-day.
@@ -61,6 +65,10 @@ type Delta struct {
Events *TopK
Dropped Dropped
Restarted bool
+ // Countries and GeoSources are pageview counts by GeoIP answer (geoip
+ // spec §9), bounded by boundedAdd.
+ Countries map[string]int64
+ GeoSources map[string]int64
// requeued marks a delta that has already been through one failed
// flush (internal/analytics.Meter.Requeue). Not serialised: it never
@@ -116,9 +124,8 @@ func (d *Day) Merge(x Delta) (bool, error) {
d.Dropped.A = addDrops(d.Dropped.A, x.Dropped.A)
d.Dropped.Flush += x.Dropped.Flush
d.Restarted = d.Restarted || x.Restarted
- if d.Countries == nil {
- d.Countries = map[string]int64{}
- }
+ d.Countries = mergeCounts(d.Countries, x.Countries, maxCountryKeys, nil)
+ d.GeoSources = mergeCounts(d.GeoSources, x.GeoSources, maxGeoSourceKeys, attributed)
d.Seq[x.Box] = x.Seq
return true, nil
}
@@ -172,3 +179,55 @@ func UpsertDay(ctx context.Context, s release.Store, sqid, app string, x Delta)
}
return fmt.Errorf("analytics: %s: lost CAS race three times", key)
}
+
+const (
+ maxCountryKeys = 300
+ maxGeoSourceKeys = 8
+)
+
+// attributed reports whether a database type carries an attribution the
+// analytics page must keep showing (geoip spec §9, §13: the DB-IP and
+// GeoLite2 prefixes). Such a source is never folded into _other, so its
+// credit survives any number of other types (codex R7-3). Kept here as two
+// prefixes rather than importing internal/geoip, which this package's
+// dependants should not need.
+func attributed(source string) bool {
+ return strings.HasPrefix(source, "DBIP-") || strings.HasPrefix(source, "GeoLite2-")
+}
+
+// boundedAdd adds n to m[k], folding a new key past max into OtherKey
+// unless keep says it must never fold.
+func boundedAdd(m map[string]int64, k string, n int64, max int, keep func(string) bool) {
+ if _, ok := m[k]; ok || k == OtherKey || len(m) < max || (keep != nil && keep(k)) {
+ m[k] += n
+ return
+ }
+ m[OtherKey] += n
+}
+
+func mergeCounts(dst, src map[string]int64, max int, keep func(string) bool) map[string]int64 {
+ if dst == nil {
+ dst = map[string]int64{}
+ }
+ for k, n := range src {
+ boundedAdd(dst, k, n, max, keep)
+ }
+ return dst
+}
+
+// countGeo applies spec §9 to one pageview: hit counts its country (or
+// _unknown for a coordinate-only answer), miss counts _unknown, off and ""
+// count nothing; a hit or miss also counts its source.
+func countGeo(d *Delta, o Obs) {
+ if o.GeoState != "hit" && o.GeoState != "miss" {
+ return
+ }
+ c := o.Country
+ if o.GeoState == "miss" || c == "" {
+ c = "_unknown"
+ }
+ boundedAdd(d.Countries, c, 1, maxCountryKeys, nil)
+ if o.GeoSource != "" {
+ boundedAdd(d.GeoSources, o.GeoSource, 1, maxGeoSourceKeys, attributed)
+ }
+}
diff --git a/internal/analytics/meter.go b/internal/analytics/meter.go
index 50d1d5baa..3c502bf5f 100644
--- a/internal/analytics/meter.go
+++ b/internal/analytics/meter.go
@@ -29,6 +29,12 @@ type Obs struct {
Prefetch bool // Sec-Purpose/Purpose: prefetch
FetchDest string // Sec-Fetch-Dest, "" when absent
When time.Time
+ // GeoIP (geoip spec §9): the request's one answer, as the ingress edge
+ // computed it or as this edge adopted it. GeoState is hit | miss | off,
+ // "" from an edge that predates GeoIP (counted like off).
+ Country string
+ GeoSource string
+ GeoState string
}
// AppKey is the (account, app) the meter buckets by.
@@ -149,7 +155,8 @@ func (m *Meter) delta(k AppKey, date string) *Delta {
if d == nil {
restarted := date == m.started && !m.restartedFlagged[k]
d = &Delta{Date: date, Box: m.box, Hours: map[string]HourRow{}, Visitors: &HLL{},
- Pages: NewTopK(), Referers: NewTopK(), Events: NewTopK(), Restarted: restarted}
+ Pages: NewTopK(), Referers: NewTopK(), Events: NewTopK(), Restarted: restarted,
+ Countries: map[string]int64{}, GeoSources: map[string]int64{}}
days[date] = d
if restarted {
m.restartedFlagged[k] = true // this app's start-day note is spent; a later drain's fresh delta for the same day must not repeat it
@@ -183,6 +190,7 @@ func (m *Meter) Observe(o Obs) {
ref = "direct"
}
d.Referers.Add(strings.ToLower(ref), 1)
+ countGeo(d, o)
}
d.Hours[hour] = row
}
diff --git a/internal/analytics/read.go b/internal/analytics/read.go
index bf80bcb68..a7fa43814 100644
--- a/internal/analytics/read.go
+++ b/internal/analytics/read.go
@@ -38,6 +38,8 @@ type Summary struct {
Pages map[string]int64
Referers map[string]int64
Events map[string]int64
+ Countries map[string]int64
+ GeoSources map[string]int64
PagesApprox bool
ReferersApprox bool
EventsApprox bool
@@ -69,6 +71,7 @@ func ReadRange(ctx context.Context, s release.Store, sqid, app string, from, to
}
sort.Strings(sum.Boxes)
pages, refs, events := NewTopK(), NewTopK(), NewTopK()
+ countries, sources := map[string]int64{}, map[string]int64{}
for _, date := range Dates(from, to) {
ds := DaySummary{Date: date}
vis := &HLL{}
@@ -98,6 +101,8 @@ func ReadRange(ctx context.Context, s release.Store, sqid, app string, from, to
pages.Merge(TopKFrom(day.Pages))
refs.Merge(TopKFrom(day.Referers))
events.Merge(TopKFrom(day.Events))
+ mergeCounts(countries, day.Countries, maxCountryKeys, nil)
+ mergeCounts(sources, day.GeoSources, maxGeoSourceKeys, attributed)
sum.Dropped.E = addDrops(sum.Dropped.E, day.Dropped.E)
sum.Dropped.A = addDrops(sum.Dropped.A, day.Dropped.A)
sum.Dropped.Flush += day.Dropped.Flush
@@ -115,6 +120,7 @@ func ReadRange(ctx context.Context, s release.Store, sqid, app string, from, to
sum.Pages, sum.PagesApprox = pages.Fold(), pages.Approximate()
sum.Referers, sum.ReferersApprox = refs.Fold(), refs.Approximate()
sum.Events, sum.EventsApprox = events.Fold(), events.Approximate()
+ sum.Countries, sum.GeoSources = countries, sources
return sum, nil
}
diff --git a/internal/console/apianalytics.go b/internal/console/apianalytics.go
index 93a0fd399..97823f5e6 100644
--- a/internal/console/apianalytics.go
+++ b/internal/console/apianalytics.go
@@ -9,6 +9,7 @@ import (
"context"
"log"
"net/http"
+ "sort"
"time"
"github.com/carlosframework/platform/internal/analytics"
@@ -33,6 +34,8 @@ type analyticsResponse struct {
Pages map[string]int64 `json:"pages"`
Referers map[string]int64 `json:"referers"`
Events map[string]int64 `json:"events"`
+ Countries map[string]int64 `json:"countries"`
+ GeoSources map[string]int64 `json:"geo_sources"`
PagesApprox bool `json:"pages_approx"`
ReferersApprox bool `json:"referers_approx"`
EventsApprox bool `json:"events_approx"`
@@ -73,7 +76,8 @@ func (s *Server) handleAPIAnalytics(w http.ResponseWriter, r *http.Request) {
PV: sum.PV, Visitors: sum.Visitors, Bytes: sum.Bytes, Bots: sum.Bots,
Days: sum.Days, Pages: sum.Pages, Referers: sum.Referers, Events: sum.Events,
PagesApprox: sum.PagesApprox, ReferersApprox: sum.ReferersApprox, EventsApprox: sum.EventsApprox,
- Dropped: sum.Dropped,
+ Dropped: sum.Dropped,
+ Countries: topWithOther(sum.Countries, 10), GeoSources: sum.GeoSources,
}
// "Nothing counted" is an empty object, not null — the same shape
// errorsResponse gives its groups, so a caller may index the answer
@@ -90,5 +94,43 @@ func (s *Server) handleAPIAnalytics(w http.ResponseWriter, r *http.Request) {
if resp.Events == nil {
resp.Events = map[string]int64{}
}
+ if resp.GeoSources == nil {
+ resp.GeoSources = map[string]int64{}
+ }
writeJSON(w, http.StatusOK, resp)
}
+
+// topWithOther keeps the n largest keys (ties by key) and sums the rest,
+// plus any existing _other, into "_other" (geoip spec §9: top 10 plus
+// other). Never nil.
+func topWithOther(m map[string]int64, n int) map[string]int64 {
+ type kv struct {
+ k string
+ v int64
+ }
+ var all []kv
+ for k, v := range m {
+ if k != "_other" {
+ all = append(all, kv{k, v})
+ }
+ }
+ sort.Slice(all, func(i, j int) bool {
+ if all[i].v != all[j].v {
+ return all[i].v > all[j].v
+ }
+ return all[i].k < all[j].k
+ })
+ out := map[string]int64{}
+ other := m["_other"]
+ for i, e := range all {
+ if i < n {
+ out[e.k] = e.v
+ } else {
+ other += e.v
+ }
+ }
+ if other > 0 {
+ out["_other"] = other
+ }
+ return out
+}
diff --git a/internal/console/apianalytics_geo_test.go b/internal/console/apianalytics_geo_test.go
new file mode 100644
index 000000000..64f3bc39b
--- /dev/null
+++ b/internal/console/apianalytics_geo_test.go
@@ -0,0 +1,26 @@
+package console
+
+import (
+ "fmt"
+ "testing"
+)
+
+func TestTopWithOther(t *testing.T) {
+ m := map[string]int64{"_other": 5}
+ for i := 0; i < 12; i++ {
+ m[fmt.Sprintf("C%02d", i)] = int64(100 - i) // C00 largest, C11 smallest
+ }
+ got := topWithOther(m, 10)
+ if len(got) != 11 {
+ t.Fatalf("len = %d, want 10 + _other: %v", len(got), got)
+ }
+ if got["_other"] != 5+89+90 { // C10=90, C11=89, plus the existing 5
+ t.Fatalf("_other = %d", got["_other"])
+ }
+ if _, ok := got["C10"]; ok {
+ t.Fatal("the 11th largest survived")
+ }
+ if len(topWithOther(nil, 10)) != 0 || topWithOther(nil, 10) == nil {
+ t.Fatal("nil in must be an empty, non-nil map out")
+ }
+}
diff --git a/internal/edge/analytics.go b/internal/edge/analytics.go
index 16c1f1df9..3c6f464a5 100644
--- a/internal/edge/analytics.go
+++ b/internal/edge/analytics.go
@@ -111,6 +111,13 @@ func (e *Edge) observeAnalytics(rt registry.Route, r *http.Request, ob *counting
if ref == nil || !e.analyticsEnabled(rt) {
return
}
+ // One observation per served response (geoip spec §9, codex R4-1): the
+ // hub that forwarded this over the tunnel already observed it, from its
+ // own cache hit or from this response.
+ if fromTunnel(r.Context()) {
+ return
+ }
+ geoAns, _ := geoAnswerFrom(r.Context())
refHost := ""
if u, err := url.Parse(r.Header.Get("Referer")); err == nil {
refHost = u.Hostname()
@@ -122,6 +129,7 @@ func (e *Edge) observeAnalytics(rt registry.Route, r *http.Request, ob *counting
Path: r.URL.Path, RefererHost: refHost, UA: r.Header.Get("User-Agent"),
ClientIP: clientIP(r), Bytes: bytes,
Prefetch: prefetch, FetchDest: strings.ToLower(r.Header.Get("Sec-Fetch-Dest")),
- When: time.Now(),
+ When: time.Now(),
+ Country: geoAns.Country, GeoSource: geoAns.Source, GeoState: string(geoAns.State),
})
}
diff --git a/internal/edge/tunnel_geo_e2e_test.go b/internal/edge/tunnel_geo_e2e_test.go
index 2847f61cc..d90930552 100644
--- a/internal/edge/tunnel_geo_e2e_test.go
+++ b/internal/edge/tunnel_geo_e2e_test.go
@@ -8,10 +8,12 @@ import (
"net/http/httptest"
"net/url"
"testing"
+ "time"
"github.com/carlosframework/platform/internal/geo"
"github.com/carlosframework/platform/internal/geoip"
"github.com/carlosframework/platform/internal/registry"
+ "github.com/carlosframework/platform/internal/release"
)
// tunnelGeoScene is a hub edge routing demo.example over a real tunnelclient
@@ -85,3 +87,86 @@ func TestTunnelCarriesAMissAsAMiss(t *testing.T) {
t.Fatal("the remote looked up after the hub missed")
}
}
+
+// Spec §9, codex R4-1/R5: one analytics observation per served response.
+// The hub observes every tunnelled response (fresh or from its own cache)
+// with its own geo answer; the remote observes none of them.
+func TestTunnelledPageviewsAreCountedOnceAtTheHub(t *testing.T) {
+ hubReg := tunnelRoute(t, registry.Route{Host: "demo.example", Kind: "instance", Status: "ready", Tunnel: "pi-box", App: "web"})
+ hub, err := New(hubReg, nil)
+ if err != nil {
+ t.Fatal(err)
+ }
+ t.Cleanup(hub.probers.Stop)
+ hub.EnableCache(NewCache(DefaultCacheMaxEntries, DefaultCacheMaxBodyBytes))
+ srv := httptest.NewServer(hub)
+ t.Cleanup(srv.Close)
+ u, _ := url.Parse(srv.URL)
+ th := NewTunnelHub(u.Hostname(), map[string]string{"pi-box": "sekrit"})
+ hub.EnableTunnels(th)
+ if err := hub.Reload(); err != nil {
+ t.Fatal(err)
+ }
+
+ page := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+ w.Header().Set("Content-Type", "text/html; charset=utf-8")
+ if r.URL.Path == "/cached" {
+ w.Header().Set("Cache-Control", "public, max-age=60")
+ }
+ w.Write([]byte("<h1>hi</h1>"))
+ }))
+ t.Cleanup(page.Close)
+ remoteReg := openReg(t)
+ if err := remoteReg.Upsert(registry.Route{Host: "demo.example", Kind: "service", Addr: page.Listener.Addr().String(), Status: "ready", App: "web"}); err != nil {
+ t.Fatal(err)
+ }
+ remote, err := New(remoteReg, nil)
+ if err != nil {
+ t.Fatal(err)
+ }
+ t.Cleanup(remote.probers.Stop)
+ stop := startFakeRemoteHost(t, srv.URL, "pi-box", "sekrit", remote)
+ t.Cleanup(stop)
+ waitForTunnelState(t, th, "pi-box", "connected", e2eReconnectBound)
+
+ hubM, remoteM := &memAnalytics{}, &memAnalytics{}
+ on := map[string]bool{release.OpsAccount + "/web": true}
+ hub.EnableAnalytics(hubM)
+ hub.SetAnalyticsOn(on)
+ remote.EnableAnalytics(remoteM)
+ remote.SetAnalyticsOn(on)
+ spyGeo(hub, "GB")
+ spyGeo(remote, "XX")
+
+ get := func(path string) {
+ req, _ := http.NewRequest("GET", srv.URL+path, nil)
+ req.Host = "demo.example"
+ req.Header.Set("User-Agent", "Mozilla/5.0")
+ resp, err := http.DefaultClient.Do(req)
+ if err != nil {
+ t.Fatal(err)
+ }
+ io.Copy(io.Discard, resp.Body)
+ resp.Body.Close()
+ }
+ get("/fresh")
+ get("/cached")
+ get("/cached") // the hub's cache answers; the remote never sees it
+
+ deadline := time.Now().Add(2 * time.Second)
+ for len(hubM.snap()) < 3 && time.Now().Before(deadline) {
+ time.Sleep(10 * time.Millisecond) // the observer runs in a defer after the response
+ }
+ if n := len(remoteM.snap()); n != 0 {
+ t.Fatalf("the remote observed %d tunnelled responses, want 0", n)
+ }
+ obs := hubM.snap()
+ if len(obs) != 3 {
+ t.Fatalf("the hub observed %d responses, want 3 (one per served response)", len(obs))
+ }
+ for _, o := range obs {
+ if o.Country != "GB" || o.GeoState != "hit" || o.GeoSource != "TestDB-GB" {
+ t.Fatalf("hub observation %+v, want its own GB answer", o)
+ }
+ }
+}