From 080141145e964945d3f105be01903b082ebbfd8b Mon Sep 17 00:00:00 2001 From: MrAlders0n <55921894+MrAlders0n@users.noreply.github.com> Date: Wed, 21 Oct 2026 11:28:55 -0400 Subject: [PATCH] Read-only scope lists, 503 on statement timeouts, log read-model restores (#264) Read-only servers now restore the saved MeshMapper scope catalogues (a read-only step), so region scope lists are no longer empty. A statement that hits the pool's statement_timeout answers 503 with Retry-After on route evidence and the read-model pages instead of 500. Read-model trust logs when it comes back after a drop, and when repair runs drop it. --- cmd/beacon/main.go | 5 +- cmd/beacon/readonly.go | 2 +- cmd/beacon/readonly_integration_test.go | 50 +++++++++++++++++++- db/readmodel.go | 9 +++- db/readmodel_test.go | 38 +++++++++++++++ docs/docs.go | 6 +++ docs/swagger.json | 6 +++ docs/swagger.yaml | 4 ++ internal/api/handlers/responses.go | 15 ++++++ internal/api/handlers/route_evidence.go | 2 + internal/api/handlers/route_evidence_test.go | 6 +++ internal/api/handlers/unavailable_metrics.go | 5 +- internal/api/handlers/unavailable_test.go | 23 +++++++++ 13 files changed, 164 insertions(+), 7 deletions(-) diff --git a/cmd/beacon/main.go b/cmd/beacon/main.go index 38a4b13..bd905d8 100644 --- a/cmd/beacon/main.go +++ b/cmd/beacon/main.go @@ -257,8 +257,9 @@ func main() { os.Exit(1) } } + // Read-only servers restore too: region scope lists need the catalogues, and the restore only reads. var scopeImporter *meshmapper.Importer - if mode.MeshMapper && cfg.MeshMapper.Scopes.Enabled { + if cfg.MeshMapper.Scopes.Enabled { restoreCtx, cancelRestore := context.WithTimeout(ctx, 10*time.Second) scopeImporter, err = meshmapper.New(restoreCtx, cfg.MeshMapper.Scopes, maintStore, directory, scopes, scopeEntries) cancelRestore() @@ -483,7 +484,7 @@ func main() { } tasks = append(tasks, background.ObserverCleanupTask(coalescer, resolved.ObserverDeleteAfter, resolved.CleanupInterval, onDelete)) } - if scopeImporter != nil { + if mode.MeshMapper && scopeImporter != nil { tasks = append(tasks, background.Task{Name: "meshmapper.scopes", Interval: meshmapper.PollInterval, Run: scopeImporter.Refresh}) } if mode.MeshMapper && cfg.MeshMapper.Zones.Enabled { diff --git a/cmd/beacon/readonly.go b/cmd/beacon/readonly.go index 593a12a..5788e64 100644 --- a/cmd/beacon/readonly.go +++ b/cmd/beacon/readonly.go @@ -22,7 +22,7 @@ type bootMode struct { Coalescer bool // Jobs runs the rollup catch-up and the background scheduler. Jobs bool - // MeshMapper restores and runs the MeshMapper importers; their restores can prune or clear rows. + // MeshMapper runs the MeshMapper importers and the restores that can prune or clear rows. MeshMapper bool } diff --git a/cmd/beacon/readonly_integration_test.go b/cmd/beacon/readonly_integration_test.go index e9875a8..bbb240b 100644 --- a/cmd/beacon/readonly_integration_test.go +++ b/cmd/beacon/readonly_integration_test.go @@ -25,6 +25,8 @@ import ( "time" "github.com/MeshCore-Beacon/beacon-server/db" + "github.com/MeshCore-Beacon/beacon-server/internal/meshmapper" + "github.com/MeshCore-Beacon/beacon-server/internal/scopestore" "github.com/jackc/pgx/v5" "github.com/jackc/pgx/v5/pgxpool" ) @@ -51,8 +53,17 @@ regions: channel_keys: hashtags: - readonlytest +%s` + +// readOnlyScopesConfig enables the scope import on the read-only boot only, so the control never reaches the network. +const readOnlyScopesConfig = `meshmapper: + scopes: + enabled: true ` +// readOnlyCatalogueScope is the only scope in the saved SEA catalogue. +const readOnlyCatalogueScope = "#rocatalogue" + // readOnlyRequests is the request mix a read-only server serves. GETs must not fail with a // server error; writes must fail and never succeed. var readOnlyRequests = []struct { @@ -161,7 +172,8 @@ func TestReadOnlyBootWritesNothingPostgres(t *testing.T) { t.Fatal("a normal boot changed no table, so the check cannot see a write") } - before = after + seedScopeCatalogue(t, ctx, cfg) + before = snapshotTables(t, ctx, statsConn(), name) log := bootAndServe(t, ctx, poolDSN(t, dsn, name), true) after = snapshotTables(t, ctx, statsConn(), name) for _, table := range slices.Sorted(maps.Keys(after)) { @@ -193,6 +205,26 @@ func TestReadOnlyBootWritesNothingPostgres(t *testing.T) { } } +// seedScopeCatalogue saves a validated MeshMapper scope catalogue for SEA, as the writer's importer would. +func seedScopeCatalogue(t *testing.T, ctx context.Context, cfg *pgxpool.Config) { + t.Helper() + pool, err := pgxpool.NewWithConfig(ctx, cfg.Copy()) + if err != nil { + t.Fatal(err) + } + defer pool.Close() + now := time.Now().UTC().Truncate(time.Second) + payload := fmt.Sprintf(`{"generated_at":%q,"region":"SEA","zones":["SEA"],"repeaters":1,"scoped":1,`+ + `"scopes":[{"name":%q,"repeaters":1,"default":0,"monitored":false,"wardriving":false}]}`, + now.Format(time.RFC3339), strings.TrimPrefix(readOnlyCatalogueScope, "#")) + cache := meshmapper.Cache{Payload: []byte(payload), CheckedAt: now, AttemptedAt: now, NextAttempt: now.Add(time.Hour)} + store := db.New(pool, pool, 0, 0) + if err := store.SaveScopeCatalogue(ctx, "SEA", "https://sea.example/get_scopes.php", cache, + []scopestore.Entry{scopestore.FromName(readOnlyCatalogueScope)}); err != nil { + t.Fatal(err) + } +} + // poolDSN points dsn at the named database. func poolDSN(t *testing.T, dsn, name string) string { t.Helper() @@ -216,7 +248,11 @@ func bootAndServe(t *testing.T, ctx context.Context, dsn string, readOnly bool) t.Helper() dir := t.TempDir() path := filepath.Join(dir, "config.yaml") - if err := os.WriteFile(path, []byte(fmt.Sprintf(readOnlyBootConfig, readOnly)), 0600); err != nil { + extra := "" + if readOnly { + extra = readOnlyScopesConfig + } + if err := os.WriteFile(path, []byte(fmt.Sprintf(readOnlyBootConfig, readOnly, extra)), 0600); err != nil { t.Fatal(err) } l, err := net.Listen("tcp", "127.0.0.1:0") @@ -299,6 +335,16 @@ func bootAndServe(t *testing.T, ctx context.Context, dsn string, readOnly bool) t.Errorf("%s %s: %d %s", r.method, r.path, resp.StatusCode, body) } } + // Region scope lists come from the restored catalogue membership. + resp, err := client.Get(base + "/api/v1/scopes?region=pnw") + if err != nil { + t.Fatal(err) + } + body, _ := io.ReadAll(resp.Body) + resp.Body.Close() + if resp.StatusCode != http.StatusOK || !strings.Contains(string(body), readOnlyCatalogueScope) { + t.Errorf("GET /api/v1/scopes?region=pnw: %d %s, want %s from the saved catalogue", resp.StatusCode, body, readOnlyCatalogueScope) + } } else { // Give the boot-time jobs a moment to write before the control stops. time.Sleep(2 * time.Second) diff --git a/db/readmodel.go b/db/readmodel.go index 4cc9b86..15d96f0 100644 --- a/db/readmodel.go +++ b/db/readmodel.go @@ -82,6 +82,7 @@ type Trust struct { mu sync.Mutex earned, dropped bool + everTrusted bool // the boot jobs log the first trust dirty, clean int gen uint64 // writer invalidations so far repairedGen uint64 // gen covered by the last finished reconcile @@ -154,8 +155,10 @@ func (t *Trust) ObserveRun(repaired int64) { } if repaired > 0 { t.dirty, t.clean = t.dirty+1, 0 - if t.dirty >= t.policy.DropAfterDirty { + if t.dirty >= t.policy.DropAfterDirty && !t.dropped { t.dropped = true + slog.Warn("read model untrusted: repair runs fixed rows the writers missed", "component", "readmodel", "model", t.model, + "repairing_runs", t.dirty, "repaired", repaired) } } else { t.dirty, t.clean = 0, t.clean+1 @@ -200,6 +203,10 @@ func (t *Trust) Repaired(gen uint64) { func (t *Trust) update() { e := t.expiry.Load() on := t.enabled && t.earned && !t.dropped && t.repairedGen >= t.gen && (e == 0 || time.Now().UnixNano() < e) + if on && !t.trusted.Load() && t.everTrusted { + slog.Info("read model trusted again: its pages are served from it", "component", "readmodel", "model", t.model) + } + t.everTrusted = t.everTrusted || on t.trusted.Store(on) v := 0.0 if on { diff --git a/db/readmodel_test.go b/db/readmodel_test.go index a2ae2b9..96accde 100644 --- a/db/readmodel_test.go +++ b/db/readmodel_test.go @@ -4,8 +4,11 @@ package db import ( + "bytes" "context" "errors" + "log/slog" + "strings" "testing" "time" @@ -316,3 +319,38 @@ func TestCmiUnseededFallbackInvalidatesAfter(t *testing.T) { t.Fatalf("generation %d after the unseeded insert, %d during it: no invalidation after the fallback", tr.Generation(), during) } } + +// Restores after a drop and policy drops are logged; the first trust is not. +func TestTrustLogsRestoresAndPolicyDrops(t *testing.T) { + var logs bytes.Buffer + prev := slog.Default() + slog.SetDefault(slog.New(slog.NewTextHandler(&logs, nil))) + t.Cleanup(func() { slog.SetDefault(prev) }) + count := func(msg string) int { + return strings.Count(logs.String(), `msg="`+msg+`" component=readmodel model=trust_log_test`) + } + const again, dropped = "read model trusted again: its pages are served from it", "read model untrusted: repair runs fixed rows the writers missed" + + tr := NewTrust("trust_log_test", true, PacketSitesTrustPolicy) + tr.Earn() + tr.Repaired(tr.Generation()) + if count(again) != 0 { + t.Fatalf("first trust logged as a restore:\n%s", logs.String()) + } + tr.Invalidate("failed insert") + tr.Repaired(tr.Generation()) + if count(again) != 1 { + t.Fatalf("restore after an invalidation logged %d times:\n%s", count(again), logs.String()) + } + tr.ObserveRun(3) + tr.ObserveRun(2) + if count(dropped) != 1 || tr.Trusted() { + t.Fatalf("policy drop logged %d times, trusted %v:\n%s", count(dropped), tr.Trusted(), logs.String()) + } + tr.ObserveRun(0) + tr.ObserveRun(0) + tr.ObserveRun(0) + if count(again) != 2 || !tr.Trusted() { + t.Fatalf("restore after clean runs logged %d restores in total, trusted %v:\n%s", count(again), tr.Trusted(), logs.String()) + } +} diff --git a/docs/docs.go b/docs/docs.go index cd7bd4d..ee038bd 100644 --- a/docs/docs.go +++ b/docs/docs.go @@ -2594,6 +2594,12 @@ const docTemplate = `{ "schema": { "$ref": "#/definitions/internal_api_handlers.APIError" } + }, + "503": { + "description": "The lookup took too long: retry after Retry-After seconds", + "schema": { + "$ref": "#/definitions/internal_api_handlers.APIError" + } } } } diff --git a/docs/swagger.json b/docs/swagger.json index 8c78774..193cc5a 100644 --- a/docs/swagger.json +++ b/docs/swagger.json @@ -2592,6 +2592,12 @@ "schema": { "$ref": "#/definitions/internal_api_handlers.APIError" } + }, + "503": { + "description": "The lookup took too long: retry after Retry-After seconds", + "schema": { + "$ref": "#/definitions/internal_api_handlers.APIError" + } } } } diff --git a/docs/swagger.yaml b/docs/swagger.yaml index a02425a..a209ceb 100644 --- a/docs/swagger.yaml +++ b/docs/swagger.yaml @@ -3640,6 +3640,10 @@ paths: description: Internal Server Error schema: $ref: '#/definitions/internal_api_handlers.APIError' + "503": + description: 'The lookup took too long: retry after Retry-After seconds' + schema: + $ref: '#/definitions/internal_api_handlers.APIError' summary: Get retained reports matching a full saved route prefix sequence tags: - Routes diff --git a/internal/api/handlers/responses.go b/internal/api/handlers/responses.go index f936342..e805e60 100644 --- a/internal/api/handlers/responses.go +++ b/internal/api/handlers/responses.go @@ -5,6 +5,7 @@ package handlers import ( + "context" "encoding/json" "errors" "fmt" @@ -14,6 +15,7 @@ import ( "github.com/MeshCore-Beacon/beacon-server/internal/api" "github.com/MeshCore-Beacon/beacon-server/internal/guards" + "github.com/jackc/pgx/v5/pgconn" ) // respondUnavailable answers a reader error the client should retry with 503 and Retry-After, in the same shape as @@ -29,10 +31,23 @@ func respondUnavailable(w http.ResponseWriter, err error) bool { guards.WriteRefusal(w, &guards.Refusal{Guard: "slow_pool", Status: http.StatusServiceUnavailable, RetryAfter: 5, Code: "service_unavailable", Message: "The server is busy with other slow requests. Retry later."}) return true + case isQueryTimeout(err): + guards.WriteRefusal(w, &guards.Refusal{Guard: "timeout", Status: http.StatusServiceUnavailable, RetryAfter: 10, + Code: "deadline_exceeded", Message: "This page took too long to load. Retry later, or ask for a shorter time range."}) + return true } return false } +// isQueryTimeout reports a statement_timeout (57014) or deadline; a client that went away is not one. +func isQueryTimeout(err error) bool { + if err == nil || errors.Is(err, context.Canceled) { + return false + } + var pgErr *pgconn.PgError + return errors.As(err, &pgErr) && pgErr.Code == "57014" || errors.Is(err, context.DeadlineExceeded) +} + // APIError is the standard error response shape for all Beacon API endpoints. // Code is a stable snake_case identifier derived from the HTTP status text, // safe to match against in client code. Message is a human-readable description. diff --git a/internal/api/handlers/route_evidence.go b/internal/api/handlers/route_evidence.go index a194306..f2938ec 100644 --- a/internal/api/handlers/route_evidence.go +++ b/internal/api/handlers/route_evidence.go @@ -90,6 +90,7 @@ func routeEvidenceQuery(r *http.Request, iata, key string, now time.Time) (api.R // @Failure 400 {object} handlers.APIError // @Failure 404 {object} handlers.APIError // @Failure 500 {object} handlers.APIError +// @Failure 503 {object} handlers.APIError "The lookup took too long: retry after Retry-After seconds" // @Router /routes/{iata}/{pathKey}/observations [get] func getRouteEvidence(reader api.Reader) http.HandlerFunc { return func(w http.ResponseWriter, r *http.Request) { @@ -109,6 +110,7 @@ func getRouteEvidence(reader api.Reader) http.HandlerFunc { respondError(w, 404, "saved route not found") case errors.Is(err, api.ErrRouteEvidenceInput): respondError(w, 400, "invalid route evidence request") + case respondUnavailable(w, err): case err != nil: respondError(w, 500, "internal server error") default: diff --git a/internal/api/handlers/route_evidence_test.go b/internal/api/handlers/route_evidence_test.go index f2268d8..1e72bf4 100644 --- a/internal/api/handlers/route_evidence_test.go +++ b/internal/api/handlers/route_evidence_test.go @@ -13,6 +13,7 @@ import ( "github.com/MeshCore-Beacon/beacon-server/internal/api" "github.com/jackc/pgx/v5" + "github.com/jackc/pgx/v5/pgconn" ) const evidenceKey = "f097439148601d9f3291c474f82fa64c" @@ -66,6 +67,8 @@ func TestRouteEvidenceHandler(t *testing.T) { {"/LONG/" + evidenceKey + "/observations", nil, 400}, {"/YOW/" + evidenceKey + "/observations", pgx.ErrNoRows, 404}, {"/YOW/" + evidenceKey + "/observations", errors.New("database unavailable"), 500}, + {"/YOW/" + evidenceKey + "/observations", fmt.Errorf("list: %w", &pgconn.PgError{Code: "57014"}), 503}, + {"/YOW/" + evidenceKey + "/observations", fmt.Errorf("list: %w", context.Canceled), 500}, } { called := false router := RoutesRouter(stubReader{getRouteEvidence: func(_ context.Context, iata, key string, q api.RouteEvidenceQuery) (*api.RouteEvidence, error) { @@ -80,6 +83,9 @@ func TestRouteEvidenceHandler(t *testing.T) { if w.Code != tc.status { t.Fatalf("%s: %d %s", tc.path, w.Code, w.Body.String()) } + if (tc.status == 503) != (w.Header().Get("Retry-After") != "") { + t.Fatalf("%v: status %d with Retry-After %q", tc.err, w.Code, w.Header().Get("Retry-After")) + } if tc.status == 400 && called { t.Fatal("invalid request reached database") } diff --git a/internal/api/handlers/unavailable_metrics.go b/internal/api/handlers/unavailable_metrics.go index 49f009a..564d76e 100644 --- a/internal/api/handlers/unavailable_metrics.go +++ b/internal/api/handlers/unavailable_metrics.go @@ -27,8 +27,11 @@ func respondUnavailableFor(w http.ResponseWriter, err error, model string) bool return false } reason := "not_ready" - if errors.Is(err, api.ErrSlowPoolBusy) { + switch { + case errors.Is(err, api.ErrSlowPoolBusy): reason = "slow_pool_busy" + case isQueryTimeout(err): + reason = "timeout" } metrics.APIUnavailable.WithLabelValues(reason, model).Inc() return true diff --git a/internal/api/handlers/unavailable_test.go b/internal/api/handlers/unavailable_test.go index f27f71f..910795d 100644 --- a/internal/api/handlers/unavailable_test.go +++ b/internal/api/handlers/unavailable_test.go @@ -6,6 +6,7 @@ package handlers import ( "context" "encoding/json" + "errors" "fmt" "net/http" "net/http/httptest" @@ -16,6 +17,7 @@ import ( "github.com/MeshCore-Beacon/beacon-server/internal/metrics" "github.com/go-chi/chi/v5" "github.com/google/uuid" + "github.com/jackc/pgx/v5/pgconn" "github.com/prometheus/client_golang/prometheus/testutil" ) @@ -110,3 +112,24 @@ func TestPagesNotReadyAnswers503(t *testing.T) { } } } + +// A statement timeout or deadline answers 503 with Retry-After and counts as timeout. +func TestQueryTimeoutAnswers503(t *testing.T) { + models := map[string]string{"region packets": "packet_sites", "messages across channels": "cmi", "channel messages": "cmi", "observer adverts": "observation_payload_types"} + before := map[string]float64{} + for _, m := range models { + before[m] = testutil.ToFloat64(metrics.APIUnavailable.WithLabelValues("timeout", m)) + } + checkUnavailable(t, &pgconn.PgError{Code: "57014", Message: "canceling statement due to statement timeout"}, "10", "deadline_exceeded") + checkUnavailable(t, context.DeadlineExceeded, "10", "deadline_exceeded") + for m, n := range map[string]float64{"packet_sites": 2, "cmi": 4, "observation_payload_types": 2} { + if got := testutil.ToFloat64(metrics.APIUnavailable.WithLabelValues("timeout", m)) - before[m]; got != n { + t.Errorf("timeout %s counted %v, want %v", m, got, n) + } + } + for _, err := range []error{nil, context.Canceled, fmt.Errorf("query: %w", context.Canceled), errors.New("boom"), &pgconn.PgError{Code: "23505"}} { + if isQueryTimeout(err) { + t.Errorf("isQueryTimeout(%v) = true", err) + } + } +}