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

@@ -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 {

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
 }
 

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)

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 {

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())
+	}
+}

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"
+                        }
                     }
                 }
             }

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"
+                        }
                     }
                 }
             }

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

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.

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:

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")
 		}

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

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)
+		}
+	}
+}