dandri/beacon-server
1
0
Fork 0
mirror of https://github.com/MeshCore-Beacon/beacon-server.git synced 2026-10-20 15:41:57 +00:00
Code Issues Packages Projects Releases Wiki Activity

feat(topology): expose cached regional scope context for 2.0

Browse Source
This commit is contained in:
n30nex
2026-10-02 08:56:35 -04:00
parent 98006a9364
commit 6ad59d3129
13 changed files with 480 additions and 10 deletions
Download Patch File Download Diff File Expand all files Collapse all files
cmd/beacon/main.go
+1
View File
@@ -425,6 +425,7 @@ func main() {
// CachedReader. Only response projections receive the geographic annotation.
reader = api.WithLocalBorders(reader, localBorders)
r := router.New(h, reader, []*ingest.Worker{broker1, broker2}, router.Options{
ScopeCatalogues: scopeImporter,
MaxConnsPerIP: resolved.MaxConnsPerIP,
MaxConnectsPerMinute: resolved.MaxConnectsPerMinute,
WSAllowedOrigins: cfg.WebSocket.AllowedOrigins,
db/nodes.go
+1
View File
@@ -127,6 +127,7 @@ func (s *Store) ListNodes(ctx context.Context, nodeType int16, iatas []string, s
Longitude: v.Longitude,
IsObserver: v.IsObserver,
ObserverID: nullableUUID(v.ObserverID),
DefaultScope: v.DefaultScopeName,
KnownNeighborCount: v.KnownNeighborCount,
NeighborIDs: v.NeighborIds,
Stale: v.LastSeen.Valid && v.LastSeen.Time.Before(time.Now().Add(-s.staleThreshold)),
db/nodes_test.go
+12
View File
@@ -706,3 +706,15 @@ func TestDeleteOldNodes(t *testing.T) {
t.Fatalf("unexpected error: %v", err)
}
}
func TestListNodes_DefaultScopePreservesCaseAndMissingValue(t *testing.T) {
ctrl := gomock.NewController(t)
mock := mockdb.NewMockQuerier(ctrl)
scope := "#YOW"
mock.EXPECT().ListNodes(gomock.Any(), gomock.Any()).Return([]sqlc.ListNodesRow{{DefaultScopeName: &scope}, {}}, nil)
store := &Store{q: mock}
page, err := store.ListNodes(context.Background(), 0, nil, nil, nil, nil, "", "", "", 0, 10, false)
if err != nil || len(page.Items) != 2 || page.Items[0].DefaultScope == nil || *page.Items[0].DefaultScope != scope || page.Items[1].DefaultScope != nil {
t.Fatalf("scope metadata lost or invented: %+v %v", page, err)
}
}
docs/docs.go
+78
View File
@@ -2223,6 +2223,29 @@ const docTemplate = `{
}
}
},
"/scope-catalogues": {
"get": {
"description": "Discovered sources only. Counts describe the source region, not individual nodes or links. No upstream request is made. An empty list means no configured importer; checkedAt=0 means no successful check. lastError or a past freshUntil marks stale metadata. Manual Beacon scopes and packet evidence are separate.",
"produces": [
"application/json"
],
"tags": [
"Scopes"
],
"summary": "Cached MeshMapper regional scope catalogues",
"responses": {
"200": {
"description": "OK",
"schema": {
"type": "array",
"items": {
"$ref": "#/definitions/github_com_MeshCore-Beacon_beacon-server_internal_meshmapper.PublicScopeCatalogue"
}
}
}
}
}
},
"/scopes": {
"get": {
"description": "Without filters, lists every stored scope name, including imported names retained after an importer is disabled or a source is removed. With IATA or region filters, lists only manual scopes configured for a matching region and imported scopes whose current MeshMapper catalogue includes a matching IATA; observed traffic does not add scopes.",
@@ -5649,6 +5672,61 @@ const docTemplate = `{
}
}
},
"github_com_MeshCore-Beacon_beacon-server_internal_meshmapper.CatalogueScope": {
"type": "object",
"properties": {
"default": {
"type": "integer"
},
"monitored": {
"type": "boolean"
},
"name": {
"type": "string"
},
"repeaters": {
"type": "integer"
},
"wardriving": {
"type": "boolean"
}
}
},
"github_com_MeshCore-Beacon_beacon-server_internal_meshmapper.PublicScopeCatalogue": {
"type": "object",
"properties": {
"checkedAt": {
"type": "integer"
},
"freshUntil": {
"type": "integer"
},
"generatedAt": {
"type": "integer"
},
"iata": {
"type": "string"
},
"lastError": {
"type": "string"
},
"repeaters": {
"type": "integer"
},
"scoped": {
"type": "integer"
},
"scopes": {
"type": "array",
"items": {
"$ref": "#/definitions/github_com_MeshCore-Beacon_beacon-server_internal_meshmapper.CatalogueScope"
}
},
"url": {
"type": "string"
}
}
},
"internal_api_handlers.APIError": {
"type": "object",
"properties": {
docs/swagger.json
+78
View File
@@ -2221,6 +2221,29 @@
}
}
},
"/scope-catalogues": {
"get": {
"description": "Discovered sources only. Counts describe the source region, not individual nodes or links. No upstream request is made. An empty list means no configured importer; checkedAt=0 means no successful check. lastError or a past freshUntil marks stale metadata. Manual Beacon scopes and packet evidence are separate.",
"produces": [
"application/json"
],
"tags": [
"Scopes"
],
"summary": "Cached MeshMapper regional scope catalogues",
"responses": {
"200": {
"description": "OK",
"schema": {
"type": "array",
"items": {
"$ref": "#/definitions/github_com_MeshCore-Beacon_beacon-server_internal_meshmapper.PublicScopeCatalogue"
}
}
}
}
}
},
"/scopes": {
"get": {
"description": "Without filters, lists every stored scope name, including imported names retained after an importer is disabled or a source is removed. With IATA or region filters, lists only manual scopes configured for a matching region and imported scopes whose current MeshMapper catalogue includes a matching IATA; observed traffic does not add scopes.",
@@ -5647,6 +5670,61 @@
}
}
},
"github_com_MeshCore-Beacon_beacon-server_internal_meshmapper.CatalogueScope": {
"type": "object",
"properties": {
"default": {
"type": "integer"
},
"monitored": {
"type": "boolean"
},
"name": {
"type": "string"
},
"repeaters": {
"type": "integer"
},
"wardriving": {
"type": "boolean"
}
}
},
"github_com_MeshCore-Beacon_beacon-server_internal_meshmapper.PublicScopeCatalogue": {
"type": "object",
"properties": {
"checkedAt": {
"type": "integer"
},
"freshUntil": {
"type": "integer"
},
"generatedAt": {
"type": "integer"
},
"iata": {
"type": "string"
},
"lastError": {
"type": "string"
},
"repeaters": {
"type": "integer"
},
"scoped": {
"type": "integer"
},
"scopes": {
"type": "array",
"items": {
"$ref": "#/definitions/github_com_MeshCore-Beacon_beacon-server_internal_meshmapper.CatalogueScope"
}
},
"url": {
"type": "string"
}
}
},
"internal_api_handlers.APIError": {
"type": "object",
"properties": {
docs/swagger.yaml
+55
View File
@@ -1575,6 +1575,42 @@ definitions:
requires_restart:
type: boolean
type: object
github_com_MeshCore-Beacon_beacon-server_internal_meshmapper.CatalogueScope:
properties:
default:
type: integer
monitored:
type: boolean
name:
type: string
repeaters:
type: integer
wardriving:
type: boolean
type: object
github_com_MeshCore-Beacon_beacon-server_internal_meshmapper.PublicScopeCatalogue:
properties:
checkedAt:
type: integer
freshUntil:
type: integer
generatedAt:
type: integer
iata:
type: string
lastError:
type: string
repeaters:
type: integer
scoped:
type: integer
scopes:
items:
$ref: '#/definitions/github_com_MeshCore-Beacon_beacon-server_internal_meshmapper.CatalogueScope'
type: array
url:
type: string
type: object
internal_api_handlers.APIError:
properties:
code:
@@ -3101,6 +3137,25 @@ paths:
summary: Search known routes by source and destination hash
tags:
- Routes
/scope-catalogues:
get:
description: Discovered sources only. Counts describe the source region, not
individual nodes or links. No upstream request is made. An empty list means
no configured importer; checkedAt=0 means no successful check. lastError or
a past freshUntil marks stale metadata. Manual Beacon scopes and packet evidence
are separate.
produces:
- application/json
responses:
"200":
description: OK
schema:
items:
$ref: '#/definitions/github_com_MeshCore-Beacon_beacon-server_internal_meshmapper.PublicScopeCatalogue'
type: array
summary: Cached MeshMapper regional scope catalogues
tags:
- Scopes
/scopes:
get:
description: Without filters, lists every stored scope name, including imported
internal/api/handlers/scope_catalogues.go
+33
View File
@@ -0,0 +1,33 @@
// Copyright 2026 Beacon Contributors
// SPDX-License-Identifier: AGPL-3.0-or-later
package handlers
import (
"net/http"
"github.com/MeshCore-Beacon/beacon-server/internal/meshmapper"
)
// ScopeCatalogueSource supplies immutable, already-cached public metadata.
type ScopeCatalogueSource interface {
Catalogues() []meshmapper.PublicScopeCatalogue
}
// ScopeCatalogues godoc
// @Summary Cached MeshMapper regional scope catalogues
// @Description Discovered sources only. Counts describe the source region, not individual nodes or links. No upstream request is made. An empty list means no configured importer; checkedAt=0 means no successful check. lastError or a past freshUntil marks stale metadata. Manual Beacon scopes and packet evidence are separate.
// @Tags Scopes
// @Produce json
// @Success 200 {array} meshmapper.PublicScopeCatalogue
// @Router /scope-catalogues [get]
func ScopeCatalogues(source ScopeCatalogueSource) http.HandlerFunc {
return func(w http.ResponseWriter, r *http.Request) {
items := []meshmapper.PublicScopeCatalogue{}
if source != nil {
items = source.Catalogues()
}
w.Header().Set("Cache-Control", "public, max-age=60")
respond(w, http.StatusOK, items)
}
}
internal/api/router/options.go
+1
View File
@@ -14,6 +14,7 @@ import (
// Options names the startup settings used by the router. Adding an optional
// route does not change New's signature or require unrelated callers to change.
type Options struct {
ScopeCatalogues handlers.ScopeCatalogueSource
MaxConnsPerIP int
MaxConnectsPerMinute int
WSAllowedOrigins []string
internal/api/router/router.go
+1
View File
@@ -109,6 +109,7 @@ func New(h *hub.Hub, reader api.Reader, workers []*ingest.Worker, opts Options)
r.Mount("/routes", handlers.RoutesRouter(reader))
r.Mount("/scopes", handlers.ScopesRouter(reader, opts.Scopes))
r.Mount("/stats", handlers.StatsRouter(reader, handlers.StatsOptions{Scopes: opts.Scopes, SeriesWindow: opts.RollupRetention}))
r.Get("/scope-catalogues", handlers.ScopeCatalogues(opts.ScopeCatalogues))
r.Mount("/traces", handlers.TracesRouter(reader))
})
internal/api/router/scope_catalogues_test.go
+28
View File
@@ -0,0 +1,28 @@
// Copyright 2026 Beacon Contributors
// SPDX-License-Identifier: AGPL-3.0-or-later
package router
import (
"net/http"
"net/http/httptest"
"strings"
"testing"
"github.com/MeshCore-Beacon/beacon-server/internal/meshmapper"
)
func TestScopeCataloguesDisabledAndReadOnly(t *testing.T) {
var importer *meshmapper.Importer
r := New(nil, nil, nil, Options{ScopeCatalogues: importer})
for _, method := range []string{http.MethodGet, http.MethodPost} {
w := httptest.NewRecorder()
r.ServeHTTP(w, httptest.NewRequest(method, "/api/v1/scope-catalogues", nil))
if method == http.MethodGet && (w.Code != 200 || strings.TrimSpace(w.Body.String()) != "[]" || w.Header().Get("Cache-Control") != "public, max-age=60") {
t.Fatal(w.Code, w.Body.String())
}
if method == http.MethodPost && w.Code != 405 {
t.Fatal(w.Code)
}
}
}
internal/meshmapper/catalogues.go
+75
View File
@@ -0,0 +1,75 @@
// Copyright 2026 Beacon Contributors
// SPDX-License-Identifier: AGPL-3.0-or-later
package meshmapper
import (
"encoding/json"
"time"
"github.com/MeshCore-Beacon/beacon-server/internal/scopestore"
)
// CatalogueScope is regional source metadata, not per-node forwarding evidence.
type CatalogueScope struct {
Name string `json:"name"`
Repeaters int `json:"repeaters"`
Default int `json:"default"`
Monitored bool `json:"monitored"`
Wardriving bool `json:"wardriving"`
}
type PublicScopeCatalogue struct {
IATA string `json:"iata"`
URL string `json:"url"`
GeneratedAt int64 `json:"generatedAt"`
CheckedAt int64 `json:"checkedAt"`
FreshUntil int64 `json:"freshUntil"`
LastError string `json:"lastError,omitempty"`
Repeaters int `json:"repeaters"`
Scoped int `json:"scoped"`
Scopes []CatalogueScope `json:"scopes"`
}
// Catalogues returns an immutable snapshot of discovered sources. No request
// triggers an upstream fetch or reads ingestion tables. Callers must not mutate it.
func (i *Importer) Catalogues() []PublicScopeCatalogue {
if i != nil {
if snapshot := i.catalogues.Load(); snapshot != nil {
return snapshot.([]PublicScopeCatalogue)
}
}
return []PublicScopeCatalogue{}
}
func (i *Importer) publishCatalogues() {
result := make([]PublicScopeCatalogue, 0, len(i.sources))
for _, s := range i.sources {
c := PublicScopeCatalogue{IATA: s.iata, URL: s.url, LastError: s.cache.LastError, Scopes: []CatalogueScope{}}
if !s.cache.CheckedAt.IsZero() {
c.CheckedAt = s.cache.CheckedAt.UnixMilli()
c.FreshUntil = s.cache.CheckedAt.Add(i.interval + 5*time.Minute).UnixMilli()
}
if len(s.cache.Payload) > 0 {
// Reuse the importer validation before exposing saved metadata. Invalid
// saved catalogues are omitted just as they are from the scope matcher.
if _, generated, err := decode(s.cache.Payload, s.iata); err == nil {
var payload struct {
Repeaters int
Scoped int
Scopes []CatalogueScope
}
if json.Unmarshal(s.cache.Payload, &payload) == nil {
c.GeneratedAt = generated.UnixMilli()
c.Repeaters, c.Scoped = payload.Repeaters, payload.Scoped
for _, scope := range payload.Scopes {
scope.Name = scopestore.FromName(scope.Name).Name
c.Scopes = append(c.Scopes, scope)
}
}
}
}
result = append(result, c)
}
i.catalogues.Store(result)
}
internal/meshmapper/catalogues_test.go
+103
View File
@@ -0,0 +1,103 @@
// Copyright 2026 Beacon Contributors
// SPDX-License-Identifier: AGPL-3.0-or-later
package meshmapper
import (
"context"
"encoding/json"
"net/http"
"net/http/httptest"
"sync"
"testing"
"time"
"github.com/MeshCore-Beacon/beacon-server/internal/config"
"github.com/MeshCore-Beacon/beacon-server/internal/scopestore"
)
func TestPublicCataloguesFreshnessAndIsolation(t *testing.T) {
status := http.StatusOK
calls := 0
server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
calls++
w.WriteHeader(status)
if status == http.StatusOK {
_, _ = w.Write(catalogue("yow", "YOW", "monitored-zero"))
}
}))
defer server.Close()
cfg := config.MeshMapperScopesConfig{Enabled: true}
store := newMemoryStore("YOW")
imp, err := New(context.Background(), cfg, store, NewDirectory(newZoneListMemory()), scopestore.New(), nil)
if err != nil {
t.Fatal(err)
}
if got := imp.Catalogues(); len(got) != 1 || got[0].CheckedAt != 0 || len(got[0].Scopes) != 0 {
t.Fatal(got)
}
imp.sources[0].url = server.URL
now := time.Now()
if err := imp.refresh(context.Background(), &imp.sources[0], now); err != nil {
t.Fatal(err)
}
imp.publishCatalogues()
old := imp.Catalogues()
if old[0].Scopes[0].Name != "#yow" || old[0].Scopes[1].Name != "#YOW" || len(old[0].Scopes) != 3 || old[0].Repeaters != 5 || old[0].Scoped != 3 || old[0].FreshUntil <= old[0].CheckedAt {
t.Fatal(old)
}
status = http.StatusServiceUnavailable
if err := imp.refresh(context.Background(), &imp.sources[0], now.Add(time.Minute)); err != nil {
t.Fatal(err)
}
imp.publishCatalogues()
current := imp.Catalogues()[0]
if current.LastError != "HTTP 503" || current.CheckedAt != old[0].CheckedAt || current.GeneratedAt != old[0].GeneratedAt || len(current.Scopes) != 3 || old[0].LastError != "" {
t.Fatal(current, old)
}
// Readers only access published immutable data; the race suite exercises concurrent refreshes.
var wg sync.WaitGroup
wg.Add(1)
go func() {
defer wg.Done()
for range 100 {
_, _ = json.Marshal(imp.Catalogues())
}
}()
for range 100 {
imp.publishCatalogues()
}
wg.Wait()
if calls != 2 {
t.Fatal("reading public metadata performed upstream requests", calls)
}
disabled, err := New(context.Background(), config.MeshMapperScopesConfig{}, store, NewDirectory(newZoneListMemory()), scopestore.New(), nil)
if err != nil || len(disabled.Catalogues()) != 0 {
t.Fatal("disabled importer exposed old membership")
}
store.rows[sourceKey{"YOW", server.URL}] = Cache{Payload: []byte(`{"scopes":"invalid"}`)}
restored, err := New(context.Background(), cfg, store, NewDirectory(newZoneListMemory()), scopestore.New(), nil)
if err != nil || len(restored.Catalogues()[0].Scopes) != 0 || restored.Catalogues()[0].LastError == "" {
t.Fatal("invalid saved metadata leaked")
}
}
func TestPublicCataloguesIncludeDiscoveredAndUnlistedRegions(t *testing.T) {
ctx := context.Background()
store := newMemoryStore("YOW")
dir := NewDirectory(newZoneListMemory())
dir.lists["CA"] = &zoneList{zones: map[string]zoneEntry{}, fetchedAt: time.Now()}
imp, err := New(ctx, config.MeshMapperScopesConfig{Enabled: true}, store, dir, scopestore.New(), nil)
if err != nil {
t.Fatal(err)
}
before := imp.Catalogues()
store.iatas = append(store.iatas, "YKF")
if err := imp.Refresh(ctx); err != nil {
t.Fatal(err)
}
after := imp.Catalogues()
if len(before) != 1 || before[0].LastError != "" || len(after) != 2 || after[1].IATA != "YKF" || after[1].LastError != "not listed" {
t.Fatal(before, after)
}
}
internal/meshmapper/scopes.go
+14 -10
View File
@@ -14,6 +14,7 @@ import (
"slices"
"strconv"
"strings"
"sync/atomic"
"time"
"unicode"
"unicode/utf8"
@@ -60,16 +61,17 @@ type source struct {
// Importer is owned by one background task; ScopeStore synchronizes its consumers.
type Importer struct {
store Store
scopes *scopestore.ScopeStore
manual []scopestore.Entry
dir *Directory
scopesURL func(site string) (string, bool)
seen map[string]bool
sources []source
interval time.Duration
client *http.Client
onChange func(context.Context)
catalogues atomic.Value // immutable []PublicScopeCatalogue, off the ingest path
store Store
scopes *scopestore.ScopeStore
manual []scopestore.Entry
dir *Directory
scopesURL func(site string) (string, bool)
seen map[string]bool
sources []source
interval time.Duration
client *http.Client
onChange func(context.Context)
}
// SetCacheInvalidator is wired once at startup, before the background task starts.
@@ -102,6 +104,7 @@ func New(ctx context.Context, cfg config.MeshMapperScopesConfig, store Store, di
}
i.track(iatas, latest)
i.publish()
i.publishCatalogues()
return i, nil
}
@@ -136,6 +139,7 @@ func (i *Importer) track(iatas []string, saved map[string]Catalogue) {
// Refresh checks only one due source. The scheduler serializes calls every 15s.
func (i *Importer) Refresh(ctx context.Context) (err error) {
defer i.publishCatalogues()
parent := ctx
defer func() {
if parent.Err() != nil {