From 55df07f9e848b22b9ea0c2e0c44ab94b52b2c04f Mon Sep 17 00:00:00 2001 From: MrAlders0n Date: Wed, 30 Sep 2026 10:28:02 -0400 Subject: [PATCH 01/13] fix(backup): abort exports on shutdown and sweep stale staging dirs srv.Shutdown never cancelled the request context, so a SIGTERM mid-export left the plaintext dump and pg_service.conf in TMPDIR. The handler now derives its context from the app context and the router removes staging dirs older than 20 minutes at startup. --- cmd/beacon/main.go | 2 +- internal/api/handlers/backup.go | 26 ++++++-- internal/api/handlers/backup_test.go | 60 +++++++++++++++++-- .../api/router/backup_integration_test.go | 2 +- internal/api/router/backup_test.go | 3 +- 5 files changed, 82 insertions(+), 11 deletions(-) diff --git a/cmd/beacon/main.go b/cmd/beacon/main.go index 24e60847..fc98fd9b 100644 --- a/cmd/beacon/main.go +++ b/cmd/beacon/main.go @@ -337,7 +337,7 @@ func main() { CORS: cfg.CORS, Server: cfg.Server, Auth: cfg.Auth, RateLimit: resolved.RateLimit, AdminRoutes: map[string]http.Handler{ "/accounts": handlers.AccountsRouter(store), - "/backup": handlers.BackupRouter(backupOpts), + "/backup": handlers.BackupRouter(backupOpts, ctx), }, }) diff --git a/internal/api/handlers/backup.go b/internal/api/handlers/backup.go index 67f9641d..e9a85fd9 100644 --- a/internal/api/handlers/backup.go +++ b/internal/api/handlers/backup.go @@ -22,12 +22,25 @@ import ( // BackupRouter returns an operator-only subrouter with one bounded export or // transfer at a time. Its caller must apply the admin authentication middleware. // The options are startup-owned; the request selects no targets or paths. -func BackupRouter(opts backup.Options) http.Handler { +// shutdown aborts an in-flight export so its staging files are removed. +func BackupRouter(opts backup.Options, shutdown context.Context) http.Handler { + sweepStaleStaging() r := chi.NewRouter() - r.Get("/", backupDownload(opts, backup.Export)) + r.Get("/", backupDownload(opts, backup.Export, shutdown)) return r } +// sweepStaleStaging removes staging dirs an earlier process left behind (SIGKILL, crash). +func sweepStaleStaging() { + matches, _ := filepath.Glob(filepath.Join(os.TempDir(), "beacon-download-*")) + cutoff := time.Now().Add(-2 * backup.DefaultTimeout) + for _, dir := range matches { + if info, err := os.Stat(dir); err == nil && info.IsDir() && info.ModTime().Before(cutoff) { + _ = os.RemoveAll(dir) + } + } +} + // backupDownload godoc // // @Summary Download a private database and saved-config backup @@ -45,7 +58,7 @@ func BackupRouter(opts backup.Options) http.Handler { // @Failure 504 {object} map[string]APIError // @Failure 507 {object} map[string]APIError // @Router /admin/backup [get] -func backupDownload(opts backup.Options, export func(context.Context, backup.Options) error) http.HandlerFunc { +func backupDownload(opts backup.Options, export func(context.Context, backup.Options) error, shutdown context.Context) http.HandlerFunc { var busy atomic.Bool return func(w http.ResponseWriter, r *http.Request) { w.Header().Set("Cache-Control", "no-store") @@ -70,11 +83,16 @@ func backupDownload(opts backup.Options, export func(context.Context, backup.Opt defer os.RemoveAll(dir) requestOpts := opts requestOpts.OutputPath = filepath.Join(dir, "backup.tar.gz") - if err := export(r.Context(), requestOpts); err != nil { + ctx, cancel := context.WithCancel(r.Context()) + defer cancel() + defer context.AfterFunc(shutdown, cancel)() + if err := export(ctx, requestOpts); err != nil { if errors.Is(err, backup.ErrTooLarge) { respondError(w, http.StatusInsufficientStorage, "backup exceeds the configured export size limit") } else if errors.Is(err, context.DeadlineExceeded) { respondError(w, 504, "backup export timed out") + } else if shutdown.Err() != nil { + respondError(w, http.StatusServiceUnavailable, "server is shutting down") } else if r.Context().Err() == nil { // Export discards pg_dump stderr and never returns connection settings. slog.Error("backup export failed", "component", "backup", "error", err) diff --git a/internal/api/handlers/backup_test.go b/internal/api/handlers/backup_test.go index 64c77247..0ad1e9cb 100644 --- a/internal/api/handlers/backup_test.go +++ b/internal/api/handlers/backup_test.go @@ -27,7 +27,7 @@ func TestBackupDownload(t *testing.T) { output = got.OutputPath return os.WriteFile(output, []byte("completed fixture"), 0600) } - handler := backupDownload(opts, export) + handler := backupDownload(opts, export, context.Background()) for _, query := range []string{"?database=other", "?path=/private", "?command=anything"} { w := httptest.NewRecorder() handler(w, httptest.NewRequest("GET", "/backup"+query, nil)) @@ -53,7 +53,7 @@ func TestBackupDownload(t *testing.T) { handler = backupDownload(opts, func(_ context.Context, got backup.Options) error { output = got.OutputPath return failure - }) + }, context.Background()) w = httptest.NewRecorder() handler(w, httptest.NewRequest("GET", "/backup", nil)) want := 500 @@ -71,7 +71,7 @@ func TestBackupDownload(t *testing.T) { } } w = httptest.NewRecorder() - BackupRouter(backup.Options{}).ServeHTTP(w, httptest.NewRequest("GET", "/", nil)) + BackupRouter(backup.Options{}, context.Background()).ServeHTTP(w, httptest.NewRequest("GET", "/", nil)) if w.Code != 503 { t.Fatal("unconfigured download enabled") } @@ -85,7 +85,7 @@ func TestBackupDownloadConcurrencyAndCancellation(t *testing.T) { close(started) <-ctx.Done() return ctx.Err() - }) + }, context.Background()) ctx, cancel := context.WithCancel(context.Background()) defer cancel() go func() { @@ -104,3 +104,55 @@ func TestBackupDownloadConcurrencyAndCancellation(t *testing.T) { t.Fatal("cancelled staging retained") } } + +func TestBackupDownloadAbortsOnShutdown(t *testing.T) { + shutdown, stop := context.WithCancel(context.Background()) + stagedCh := make(chan string, 1) + handler := backupDownload(backup.Options{ConnectionService: "fixture"}, func(ctx context.Context, opts backup.Options) error { + stagedCh <- filepath.Dir(opts.OutputPath) + <-ctx.Done() + return ctx.Err() + }, shutdown) + rec := httptest.NewRecorder() + done := make(chan struct{}) + go func() { + handler.ServeHTTP(rec, httptest.NewRequest("GET", "/", nil)) + close(done) + }() + staged := <-stagedCh + stop() + select { + case <-done: + case <-time.After(2 * time.Second): + t.Fatal("export did not stop on shutdown") + } + if rec.Code != http.StatusServiceUnavailable { + t.Fatalf("status %d, want 503", rec.Code) + } + if _, err := os.Stat(staged); !os.IsNotExist(err) { + t.Fatalf("staging dir survived shutdown: %v", err) + } +} + +func TestBackupRouterSweepsStaleStaging(t *testing.T) { + t.Setenv("TMPDIR", t.TempDir()) + stale, err := os.MkdirTemp("", "beacon-download-") + if err != nil { + t.Fatal(err) + } + old := time.Now().Add(-3 * backup.DefaultTimeout) + if err := os.Chtimes(stale, old, old); err != nil { + t.Fatal(err) + } + fresh, err := os.MkdirTemp("", "beacon-download-") + if err != nil { + t.Fatal(err) + } + BackupRouter(backup.Options{}, context.Background()) + if _, err := os.Stat(stale); !os.IsNotExist(err) { + t.Fatal("stale staging dir not swept") + } + if _, err := os.Stat(fresh); err != nil { + t.Fatal("fresh staging dir swept") + } +} diff --git a/internal/api/router/backup_integration_test.go b/internal/api/router/backup_integration_test.go index 33965a4b..18f61830 100644 --- a/internal/api/router/backup_integration_test.go +++ b/internal/api/router/backup_integration_test.go @@ -91,7 +91,7 @@ func TestBackupDownloadPostgres(t *testing.T) { Auth: config.AuthConfig{APIKey: "fixture-key"}, AdminRoutes: map[string]http.Handler{"/backup": handlers.BackupRouter(backup.Options{ ConnectionService: service, ConfigPath: configPath, MaxBytes: 16 << 20, Timeout: 30 * time.Second, Version: "http-fixture", - })}, + }, context.Background())}, })) defer server.Close() request, _ := http.NewRequestWithContext(ctx, "GET", server.URL+"/api/v1/admin/backup", nil) diff --git a/internal/api/router/backup_test.go b/internal/api/router/backup_test.go index 76bdf6e7..2c026529 100644 --- a/internal/api/router/backup_test.go +++ b/internal/api/router/backup_test.go @@ -4,6 +4,7 @@ package router import ( + "context" "net/http" "net/http/httptest" "testing" @@ -15,7 +16,7 @@ import ( func TestBackupAuthenticationBeforeExport(t *testing.T) { for _, key := range []string{"", "fixture-key"} { - handler := New(nil, nil, nil, Options{Auth: config.AuthConfig{APIKey: key}, AdminRoutes: map[string]http.Handler{"/backup": handlers.BackupRouter(backup.Options{ConnectionService: "fixture"})}}) + handler := New(nil, nil, nil, Options{Auth: config.AuthConfig{APIKey: key}, AdminRoutes: map[string]http.Handler{"/backup": handlers.BackupRouter(backup.Options{ConnectionService: "fixture"}, context.Background())}}) for _, token := range []string{"", "wrong", "fixture-key"} { r := httptest.NewRequest("GET", "/api/v1/admin/backup?database=other", nil) if token != "" { From 4a9e578259958b0a645d6e419a2394b7f991c4d5 Mon Sep 17 00:00:00 2001 From: MrAlders0n Date: Wed, 30 Sep 2026 10:28:03 -0400 Subject: [PATCH 02/13] fix(ingest): record hearing paths right after the insert MarkSent ran after the side effects, so a broker copy on another worker could win it and be streamed as a repeat of the first hearing. It now runs before any further DB work, unconditionally so the set is warm before anyone opts in, and never for TRACE, whose path bytes are per-hop SNR. --- internal/api/packets.go | 2 +- internal/hub/hub.go | 15 ++++--- internal/hub/sent.go | 2 +- internal/ingest/ingest_test.go | 4 ++ internal/ingest/packet.go | 12 +++--- internal/ingest/repeats_test.go | 69 +++++++++++++++++++++++++++++++++ 6 files changed, 88 insertions(+), 16 deletions(-) diff --git a/internal/api/packets.go b/internal/api/packets.go index 2b2e4ebf..e5f47fe8 100644 --- a/internal/api/packets.go +++ b/internal/api/packets.go @@ -40,7 +40,7 @@ type PacketSummary struct { LastHeardAt int64 `json:"lastHeardAt"` // epoch ms ObservationCount int32 `json:"observationCount"` LatestObserver *PacketLatestObserver `json:"latestObserver,omitempty"` - Summary *string `json:"summary,omitempty"` // advert name from this packet; omitted when unavailable or unsupported + Summary *string `json:"summary,omitempty"` // advert name, or the ACK/TRACE/PING summary; omitted when unavailable } // PacketPathLength is the decoded path_length byte from a packet observation. diff --git a/internal/hub/hub.go b/internal/hub/hub.go index 09c22077..28061264 100644 --- a/internal/hub/hub.go +++ b/internal/hub/hub.go @@ -38,11 +38,10 @@ const ( // Event is a single fan-out unit. Payload is pre-serialised JSON so the // broadcast loop never touches encoding — it's done once by the ingest path. // -// PayloadResolved is an optional second serialization carrying additional -// fields for clients that opted into them via configure (currently just -// resolvedPath on packetObservation events). Left nil for event types that -// don't have an opt-in variant; the hub falls back to Payload in that case. -// The WithKey variants also carry observerPublicKey. +// PayloadResolved/PayloadWithKey/PayloadResolvedWithKey are optional variants +// for clients that opted in via configure (resolvedPath, observerPublicKey); +// nil falls back to Payload. Repeat marks later hearings that only +// includeRepeats clients receive. type Event struct { Type EventType Payload json.RawMessage @@ -239,13 +238,13 @@ func (h *Hub) ObserverKeyWanted() bool { return h.observerKeyClients.Load() > 0 } -// RepeatsWanted lets ingest skip repeat work when nobody wants it. +// RepeatsWanted lets ingest skip broadcasting repeats when nobody wants them. func (h *Hub) RepeatsWanted() bool { return h.repeatClients.Load() > 0 } -// MarkSent records a hearing's path and reports whether it was not already sent recently, -// so broker copies and same-path duplicates go out once. +// MarkSent records every hearing's path and reports whether it is new within the TTL, +// so broker copies and same-path duplicates are never streamed as repeats. func (h *Hub) MarkSent(packetHash, observerID, path []byte) bool { return h.sent.mark(packetHash, observerID, path, time.Now()) } diff --git a/internal/hub/sent.go b/internal/hub/sent.go index f1e04112..45ac8015 100644 --- a/internal/hub/sent.go +++ b/internal/hub/sent.go @@ -14,7 +14,7 @@ const ( sentPathsMax = 200_000 ) -// sentPaths remembers recently sent (packet, observer, path) hearings. It is shared by all +// sentPaths remembers recently heard (packet, observer, path) hearings. It is shared by all // broker workers; keys are hashed so memory stays small at the size bound. type sentPaths struct { mu sync.Mutex diff --git a/internal/ingest/ingest_test.go b/internal/ingest/ingest_test.go index 248d3a4b..aa8f892a 100644 --- a/internal/ingest/ingest_test.go +++ b/internal/ingest/ingest_test.go @@ -184,6 +184,7 @@ type stubDB struct { observationInserted bool insertChannelMessageResult bool // configurable return for InsertChannelMessage; default false undecryptedPackets []UndecryptedPacket + dbHook func() // runs in UpsertChannelIATA, the first DB call after InsertObservation } type setCapabilityCall struct { @@ -279,6 +280,9 @@ func (s *stubDB) ListUndecryptedGroupTextPackets(_ context.Context) ([]Undecrypt } func (s *stubDB) UpsertChannelIATA(_ context.Context, _ []byte, _ string, _ time.Time) error { + if s.dbHook != nil { + s.dbHook() + } s.upsertChannelIATACalls++ return nil } diff --git a/internal/ingest/packet.go b/internal/ingest/packet.go index fda54975..ad5336eb 100644 --- a/internal/ingest/packet.go +++ b/internal/ingest/packet.go @@ -81,7 +81,7 @@ type packetObservationEvent struct { IsFirstObservation bool `json:"isFirstObservation"` ObservationCount int64 `json:"observationCount"` Scope *string `json:"scope,omitempty"` - Summary *string `json:"summary,omitempty"` // same advert name as REST list/backfill rows + Summary *string `json:"summary,omitempty"` // advert name, or "ACK …"/"TRACE …"/"PING …" for those types // Only set on later hearings sent to includeRepeats clients; those carry observationCount 0. IsRepeat bool `json:"isRepeat,omitempty"` } `json:"packet"` @@ -780,6 +780,11 @@ func (w *Worker) handlePacket(ctx context.Context, iata, pubkeyHex string, raw [ w.log.Error(fmt.Sprintf("db: insert observation failed from %s/%s", iata, pubkeyHex), "error", err) return } + // Record the path before any more DB work so a broker copy racing through it can't be + // streamed as a repeat of this hearing; TRACE reuses Path for per-hop SNR. + pathNew := packet.PayloadType() != meshcore.PayloadTypeTrace && w.hub.MarkSent(packetHash[:], id[:], packet.Path) + // A duplicate is streamed, never stored, to includeRepeats clients when its path is new. + repeat := !inserted && pathNew && w.hub.RepeatsWanted() if scopeID != nil && inserted { if err := w.db.UpsertObserverScope(ctx, id, *scopeID); err != nil { @@ -841,14 +846,9 @@ func (w *Worker) handlePacket(ctx context.Context, iata, pubkeyHex string, raw [ } w.runCapabilityDetection(ctx, packet.PayloadType(), packet.PathHashSize(), resolvedIDs) - // A duplicate is streamed, never stored, to includeRepeats clients when its path is new. - repeat := !inserted && w.hub.RepeatsWanted() && w.hub.MarkSent(packetHash[:], id[:], packet.Path) if inserted || repeat { if inserted { w.handlePayloadTypeSideEffects(ctx, packet, iata, packetHash[:], radio, scopeID, matchedScope, pubkeyBytes, float32(parseNumber(envelope.SNR))) - if w.hub.RepeatsWanted() { - w.hub.MarkSent(packetHash[:], id[:], packet.Path) // so broker copies of it aren't repeats - } } // Resolve after advert updates; suppressed copies need no endpoint lookup. var resolvedSource, resolvedDestination *api.ResolvedHop diff --git a/internal/ingest/repeats_test.go b/internal/ingest/repeats_test.go index 5766c099..0f9e73d9 100644 --- a/internal/ingest/repeats_test.go +++ b/internal/ingest/repeats_test.go @@ -11,6 +11,7 @@ import ( "time" "github.com/MeshCore-Beacon/beacon-server/internal/hub" + "github.com/google/uuid" ) type countQueryDB struct { @@ -130,3 +131,71 @@ func TestHandlePacket_RepeatsOffByDefault(t *testing.T) { t.Fatalf("duplicate ran a count query (%d total)", r.db.countQueries) } } + +// A broker copy racing through the inserted hearing's DB work must find its path already +// recorded, otherwise it would be streamed as a repeat of the first hearing. +func TestHandlePacket_PathMarkedRightAfterInsert(t *testing.T) { + r := newRepeatHarness(t, true) + packet := buildGrpTxtPacket(t, 0x1a, make([]byte, 16)) + packet.Path = []byte{0x11} + packet.PathLength = 1 + hash := packet.PacketHash() + var fresh bool + r.db.dbHook = func() { + // stubDB.UpsertObserver returns uuid.Nil, so that is the observer id ingest marks with + fresh = r.w.hub.MarkSent(hash[:], uuid.Nil[:], packet.Path) + } + r.db.observationInserted = true + r.w.handlePacket(r.ctx, "YOW", "0102", packetEnvelope(r.t, packet)) + if fresh { + t.Fatal("path was not marked right after the insert") + } +} + +func TestHandlePacket_PathsRecordedWithoutSubscribers(t *testing.T) { + r := newRepeatHarness(t, false) + if got := r.hear(true, 0x11); len(got) != 1 { + t.Fatalf("first hearing: %s", got) + } + r.w.hub.Configure(r.client, hub.ClientOptions{IncludeRepeats: true}) + for !r.w.hub.RepeatsWanted() { + if r.ctx.Err() != nil { + t.Fatal("repeats opt-in never registered") + } + time.Sleep(time.Millisecond) + } + if got := r.hear(false, 0x11); len(got) != 0 { + t.Fatalf("same path streamed as repeat after late opt-in: %s", got) + } +} + +func TestHandlePacket_TraceNeverRepeats(t *testing.T) { + r := newRepeatHarness(t, true) + packet := buildTracePacket(t) + packet.Path = []byte{0x05, 0x06} + packet.PathLength = 2 + r.db.observationInserted = true + r.w.handlePacket(r.ctx, "YOW", "0102", packetEnvelope(r.t, packet)) + r.db.observationInserted = false + packet.Path = append([]byte{}, packet.Path...) + packet.Path[0] ^= 0xff // a different per-hop SNR byte, same route + r.w.handlePacket(r.ctx, "YOW", "0102", packetEnvelope(r.t, packet)) + r.w.hub.Broadcast(hub.Event{Type: hub.EventObserverStatus}) + repeats := 0 + for { + select { + case evt := <-r.client.Send: + if evt.Type == hub.EventObserverStatus { + if repeats != 0 { + t.Fatalf("TRACE copy streamed as %d repeat(s)", repeats) + } + return + } + if strings.Contains(string(evt.Payload), "isRepeat") { + repeats++ + } + case <-r.ctx.Done(): + t.Fatal("marker event not delivered") + } + } +} From cccb78f10f13b17554427377a83e44c2ae07828d Mon Sep 17 00:00:00 2001 From: MrAlders0n Date: Wed, 30 Sep 2026 10:28:03 -0400 Subject: [PATCH 03/13] fix(ingest): let paho auto-ack MQTT deliveries Sessions are clean and client IDs random, so deferring acks never bought redelivery; it only capped delivery at the broker's inflight window and pushed backlog into the broker queue, where drops are silent. --- internal/ingest/ingest.go | 8 ++------ internal/ingest/queue_test.go | 15 ++------------- 2 files changed, 4 insertions(+), 19 deletions(-) diff --git a/internal/ingest/ingest.go b/internal/ingest/ingest.go index 289a0def..dcbe2bb8 100644 --- a/internal/ingest/ingest.go +++ b/internal/ingest/ingest.go @@ -249,7 +249,6 @@ func (w *Worker) Start(ctx context.Context) { } queue := newMessageQueue(8, 2048, 32<<20, func(ctx context.Context, msg mqtt.Message) { w.handleMessageContext(ctx, msg) - msg.Ack() }) reportDrops := func() { if n := queue.takeDropped(); n > 0 { @@ -268,13 +267,13 @@ func (w *Worker) Start(ctx context.Context) { // Isolate workers across deployments; Paho reuses this ID on reconnect. // Keep it alphanumeric and within MQTT 3.1's 23-character client ID limit. + // Auto-ack: sessions are clean, so deferring acks only throttled the broker's inflight window. opts := mqtt.NewClientOptions(). AddBroker(w.cfg.URL). SetClientID(rand.Text()[:23]). SetUsername(w.cfg.Username). SetPassword(w.cfg.Password). SetAutoReconnect(true). - SetAutoAckDisabled(true). SetMaxReconnectInterval(30 * time.Second). SetKeepAlive(30 * time.Second). SetPingTimeout(10 * time.Second). @@ -335,10 +334,7 @@ func (w *Worker) subscribe(client mqtt.Client, queue *messageQueue) { // meshcore/{IATA}/{pubkey}/status // We do NOT subscribe to /internal (Role 2 access). tok := client.Subscribe("meshcore/#", 1, func(_ mqtt.Client, msg mqtt.Message) { - if !queue.enqueue(msg) { - // Counted drops must release the broker's inflight slot. - msg.Ack() - } + queue.enqueue(msg) }) if tok.Wait() && tok.Error() != nil { w.log.Error("subscribe error", "error", tok.Error()) diff --git a/internal/ingest/queue_test.go b/internal/ingest/queue_test.go index 130a17a2..6b43fe49 100644 --- a/internal/ingest/queue_test.go +++ b/internal/ingest/queue_test.go @@ -141,24 +141,13 @@ type completedSubscribe struct{ mqtt.Token } func (completedSubscribe) Wait() bool { return true } func (completedSubscribe) Error() error { return nil } -type ackMessage struct { - mqtt.Message - acked bool -} - -func (m *ackMessage) Ack() { m.acked = true } - -func TestSubscribeReleasesRejectedMessage(t *testing.T) { +func TestSubscribeCountsRejectedMessage(t *testing.T) { w, _ := newTestWorker() q := newMessageQueue(1, 1, 1, func(context.Context, mqtt.Message) { t.Error("rejected message processed") }) defer q.close(context.Background()) client := &subscribeCapture{} w.subscribe(client, q) - m := &ackMessage{Message: queueMessage("aa", "packets", 0)} - client.callback(client, m) - if !m.acked { - t.Fatal("rejected message still occupies the broker inflight window") - } + client.callback(client, queueMessage("aa", "packets", 0)) if q.takeDropped() != 1 { t.Fatal("rejected message was not counted") } From 3d7d68dfdf6e58ca1a0f106a35c068465b6a65fd Mon Sep 17 00:00:00 2001 From: MrAlders0n Date: Wed, 30 Sep 2026 10:28:03 -0400 Subject: [PATCH 04/13] ci: run packet summary and reconfirm tests, read analytics on 039 views --- .github/workflows/ci.yml | 2 +- db/analytics_retention_integration_test.go | 18 ++++++++++++++++++ 2 files changed, 19 insertions(+), 1 deletion(-) diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index 328b4528..9574594d 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -58,7 +58,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|PacketEndpointsResolveLive|ObserverMetrics|RouteEvidence|RouteEvidenceIndex|AnalyticsRetention|AnalyticsRetentionConcurrent|DeleteOldPacketsBatches|MeshMapperCatalogue|ChannelMessageScopes)Postgres$' -count=1 -v + run: go test ./db -run '^Test(Signal|Paths|PacketSummaries|PacketEndpointsResolveLive|ObserverMetrics|RouteEvidence|RouteEvidenceIndex|AnalyticsRetention|AnalyticsRetentionConcurrent|DeleteOldPacketsBatches|MeshMapperCatalogue|ChannelMessageScopes|Reconfirm[A-Za-z]*)Postgres$' -count=1 -v - name: Verify backup command against PostgreSQL 16 run: | diff --git a/db/analytics_retention_integration_test.go b/db/analytics_retention_integration_test.go index 10e33339..2bfbdfe6 100644 --- a/db/analytics_retention_integration_test.go +++ b/db/analytics_retention_integration_test.go @@ -167,6 +167,24 @@ func TestAnalyticsRetentionPostgres(t *testing.T) { t.Fatalf("missing unfiltered time index %s: %s %v", name, definition, err) } } + // The signal/path stores must read the 039 view definitions, not just 035/036. + store := &Store{q: sqlc.New(tx)} + hour := time.Now().UTC().Truncate(time.Hour) + signal, err := store.GetSignalStats(ctx, hour.Add(-6*24*time.Hour), hour, nil) + if err != nil { + t.Fatal(err) + } + // 12 observations; only the n=1 rows carry a signal (snr 0, rssi -100). + if signal.Receptions != 12 || signal.SNR.Samples != 4 || signal.RSSI.Samples != 4 { + t.Fatalf("signal stats on 039 views: %+v", signal) + } + paths, err := store.GetPathStats(ctx, hour.Add(-6*24*time.Hour), hour, nil) + if err != nil { + t.Fatal(err) + } + if paths.Receptions != 12 { + t.Fatalf("path stats on 039 views: %+v", paths) + } before := analyticsSnapshot(t, ctx, tx) q := sqlc.New(tx) cutoff := time.Now().Add(-72 * time.Hour) From c63945559898cf80753d8171ceb79151ceef8050 Mon Sep 17 00:00:00 2001 From: MrAlders0n Date: Wed, 30 Sep 2026 10:28:03 -0400 Subject: [PATCH 05/13] fix(api): log the cause of account 500s --- internal/api/handlers/accounts.go | 2 ++ internal/api/handlers/accounts_test.go | 17 +++++++++++++++++ 2 files changed, 19 insertions(+) diff --git a/internal/api/handlers/accounts.go b/internal/api/handlers/accounts.go index 92c1fc20..3f4fb287 100644 --- a/internal/api/handlers/accounts.go +++ b/internal/api/handlers/accounts.go @@ -7,6 +7,7 @@ import ( "encoding/json" "errors" "io" + "log/slog" "mime" "net/http" @@ -168,6 +169,7 @@ func accountError(w http.ResponseWriter, err error) { case errors.Is(err, api.ErrAccountInactive): respondError(w, 409, api.ErrAccountInactive.Error()) default: + slog.Error("account request failed", "component", "api", "error", err) respondError(w, 500, "internal server error") } } diff --git a/internal/api/handlers/accounts_test.go b/internal/api/handlers/accounts_test.go index d6b96833..50aaa090 100644 --- a/internal/api/handlers/accounts_test.go +++ b/internal/api/handlers/accounts_test.go @@ -4,10 +4,12 @@ package handlers import ( + "bytes" "context" "encoding/json" "errors" "fmt" + "log/slog" "net/http" "net/http/httptest" "strings" @@ -145,3 +147,18 @@ func TestAccountListAndAuthorization(t *testing.T) { } } } + +func TestAccountErrorLogsCause(t *testing.T) { + var buf bytes.Buffer + prev := slog.Default() + slog.SetDefault(slog.New(slog.NewTextHandler(&buf, nil))) + t.Cleanup(func() { slog.SetDefault(prev) }) + rec := httptest.NewRecorder() + accountError(rec, errors.New("private-error")) + if rec.Code != 500 || strings.Contains(rec.Body.String(), "private-error") { + t.Fatalf("response leaked or wrong status: %d %s", rec.Code, rec.Body.String()) + } + if !strings.Contains(buf.String(), "private-error") { + t.Fatalf("cause not logged: %s", buf.String()) + } +} From 9400024c0980a73948a9a38c3520004323ef4873 Mon Sep 17 00:00:00 2001 From: MrAlders0n Date: Wed, 30 Sep 2026 10:28:03 -0400 Subject: [PATCH 06/13] fix(profiling): offset periodic captures and harden the recorder Periodic ticks now start five minutes off the task tickers so the hourly reconfirm trigger stops colliding with the cooldown, any capture clears a deferred periodic, subdirectories no longer stop the recorder, and the metadata sidecar is published before the profile. --- PROFILING.md | 9 ++-- internal/profiling/profiling.go | 41 ++++++++++++++---- internal/profiling/profiling_test.go | 65 ++++++++++++++++++++++++++-- 3 files changed, 99 insertions(+), 16 deletions(-) diff --git a/PROFILING.md b/PROFILING.md index 5e3667ee..a9989b4a 100644 --- a/PROFILING.md +++ b/PROFILING.md @@ -31,17 +31,20 @@ can enable a longer session. Profiling adds overhead while a capture is active. ## Capture behavior -- One 30-second sample immediately, then every 30 minutes. +- One 30-second sample immediately, then every 30 minutes, offset five minutes past + the half hour so a maintenance trigger due on the hour is not lost to the cooldown. - Route reconfirmation requests an additional sample when the maintenance task starts. This includes the retention step before route validation. A five-minute cooldown between capture starts prevents overlap and repeated triggers from increasing load; - a periodic sample due during the cooldown runs when it ends. + a periodic sample due during the cooldown runs when it ends. A triggered sample + during the cooldown is skipped, and any sample cancels a periodic one waiting on + the cooldown. - Background task stacks carry a `task` label while profiling is enabled. - Shutdown or expiry stops the active sample and saves the shorter profile. - Each profile is limited to 8 MiB. The dedicated directory is limited to 256 MiB and 512 files, including metadata and files left by interrupted runs. The recorder reserves space for a full capture before starting and stops when a limit is reached. - Files are never automatically deleted. Existing unrelated files consume the budget. + Files are never automatically deleted. Existing unrelated regular files consume the budget; subdirectories are ignored. - Profiles and metadata are written with mode `0600`. A `.partial` file indicates an interrupted capture and is not a completed profile. diff --git a/internal/profiling/profiling.go b/internal/profiling/profiling.go index ca836d9d..0d61a933 100644 --- a/internal/profiling/profiling.go +++ b/internal/profiling/profiling.go @@ -10,6 +10,7 @@ import ( "errors" "fmt" "io" + "io/fs" "log/slog" "os" "path/filepath" @@ -155,8 +156,12 @@ func (r *Recorder) run(ctx context.Context) { } func schedule(ctx context.Context, requests <-chan string, capture func(context.Context, string) error) error { - ticker := time.NewTicker(captureInterval) - defer ticker.Stop() + // Offset periodic ticks from the task tickers started alongside the recorder, so a + // task's trigger isn't swallowed by a periodic capture that began seconds earlier. + first := time.NewTimer(captureInterval + captureCooldown) + defer first.Stop() + var ticker *time.Ticker + tick := first.C reason := "periodic" var last time.Time var deferred <-chan time.Time @@ -164,8 +169,9 @@ func schedule(ctx context.Context, requests <-chan string, capture func(context. if ctx.Err() != nil { return nil } - if wait := captureCooldown - time.Since(last); last.IsZero() || wait <= 0 { + if wait := captureCooldown - time.Since(last); wait <= 0 { last = time.Now() + deferred = nil // this sample covers any periodic one waiting on the cooldown if err := capture(ctx, reason); err != nil { return err } @@ -175,8 +181,15 @@ func schedule(ctx context.Context, requests <-chan string, capture func(context. } select { case <-ctx.Done(): + if ticker != nil { + ticker.Stop() + } return nil - case <-ticker.C: + case <-tick: + if ticker == nil { + ticker = time.NewTicker(captureInterval) + tick = ticker.C + } reason = "periodic" case <-deferred: deferred = nil @@ -211,11 +224,14 @@ func (r *Recorder) checkBudget() error { var total int64 for _, entry := range entries { info, err := entry.Info() + if errors.Is(err, fs.ErrNotExist) { + continue // removed between ReadDir and Info + } if err != nil { return err } if !info.Mode().IsRegular() { - return errors.New("profiling directory contains a non-regular file") + continue // subdirectories and specials don't hold profiles } total += info.Size() } @@ -260,9 +276,6 @@ func (r *Recorder) capture(ctx context.Context, reason string) error { if closeErr != nil { return closeErr } - if err := r.root.Rename(partial, name); err != nil { - return err - } data, err := json.MarshalIndent(m, "", " ") if err != nil { return err @@ -270,10 +283,12 @@ func (r *Recorder) capture(ctx context.Context, reason string) error { if len(data) > maxMetadataBytes { return errors.New("profiling metadata budget exhausted") } - meta, err := r.root.OpenFile(name+".json", os.O_WRONLY|os.O_CREATE|os.O_EXCL, 0600) + metaPartial := name + ".json.partial" + meta, err := r.root.OpenFile(metaPartial, os.O_WRONLY|os.O_CREATE|os.O_EXCL, 0600) if err != nil { return err } + defer r.root.Remove(metaPartial) _, err = meta.Write(data) closeErr = meta.Close() if err != nil { @@ -282,6 +297,14 @@ func (r *Recorder) capture(ctx context.Context, reason string) error { if closeErr != nil { return closeErr } + // Sidecar first: a .pprof without .json would look complete but be unexplained. + if err := r.root.Rename(metaPartial, name+".json"); err != nil { + return err + } + if err := r.root.Rename(partial, name); err != nil { + r.root.Remove(name + ".json") + return err + } slog.Info("CPU profile saved", "component", "profiling", "file", name, "reason", reason, "duration", m.FinishedAt.Sub(started)) return nil } diff --git a/internal/profiling/profiling_test.go b/internal/profiling/profiling_test.go index 4a307db6..51b490de 100644 --- a/internal/profiling/profiling_test.go +++ b/internal/profiling/profiling_test.go @@ -251,13 +251,13 @@ func TestScheduleBoundsExtraCaptures(t *testing.T) { time.Sleep(4 * time.Minute) requests <- "reconfirm" synctest.Wait() - time.Sleep(56 * time.Minute) + time.Sleep(61 * time.Minute) cancel() synctest.Wait() if err := <-done; err != nil { t.Fatal(err) } - want := []event{{0, "periodic"}, {5 * time.Minute, "reconfirm"}, {30 * time.Minute, "periodic"}, {60 * time.Minute, "periodic"}} + want := []event{{0, "periodic"}, {5 * time.Minute, "reconfirm"}, {35 * time.Minute, "periodic"}, {65 * time.Minute, "periodic"}} if !reflect.DeepEqual(events, want) { t.Fatalf("captures %+v, want %+v", events, want) } @@ -282,7 +282,7 @@ func TestScheduleDefersPeriodicCaptureInCooldown(t *testing.T) { }) }() synctest.Wait() - time.Sleep(28 * time.Minute) + time.Sleep(33 * time.Minute) requests <- "reconfirm" synctest.Wait() time.Sleep(33 * time.Minute) @@ -291,7 +291,7 @@ func TestScheduleDefersPeriodicCaptureInCooldown(t *testing.T) { if err := <-done; err != nil { t.Fatal(err) } - wantAt := []time.Duration{0, 28 * time.Minute, 33 * time.Minute, 60 * time.Minute} + wantAt := []time.Duration{0, 33 * time.Minute, 38 * time.Minute, 65 * time.Minute} wantReasons := []string{"periodic", "reconfirm", "periodic", "periodic"} if !reflect.DeepEqual(at, wantAt) || !reflect.DeepEqual(reasons, wantReasons) { t.Fatalf("captures %v %v, want %v %v", at, reasons, wantAt, wantReasons) @@ -299,6 +299,63 @@ func TestScheduleDefersPeriodicCaptureInCooldown(t *testing.T) { }) } +func TestScheduleOneCaptureCoversDeferredAndTrigger(t *testing.T) { + synctest.Test(t, func(t *testing.T) { + ctx, cancel := context.WithCancel(context.Background()) + defer cancel() + requests := make(chan string, 1) + var at []time.Duration + start := time.Now() + done := make(chan error, 1) + go func() { + done <- schedule(ctx, requests, func(ctx context.Context, reason string) error { + at = append(at, time.Since(start)) + time.Sleep(30 * time.Second) + return nil + }) + }() + synctest.Wait() + time.Sleep(33 * time.Minute) + requests <- "reconfirm" // 33m: captured; the 35m tick is deferred to 38m + synctest.Wait() + time.Sleep(5 * time.Minute) // 38m: the deferred periodic and a fresh trigger are both due + requests <- "reconfirm" + synctest.Wait() + time.Sleep(28 * time.Minute) // past the 65m tick so cancel can't race it + cancel() + synctest.Wait() + if err := <-done; err != nil { + t.Fatal(err) + } + want := []time.Duration{0, 33 * time.Minute, 38 * time.Minute, 65 * time.Minute} + if !reflect.DeepEqual(at, want) { + t.Fatalf("captures %v, want %v", at, want) + } + }) +} + +func TestBudgetIgnoresNonRegularEntries(t *testing.T) { + requireSupportedPlatform(t) + dir := filepath.Join(t.TempDir(), "profiles") + if err := os.MkdirAll(filepath.Join(dir, "lost+found"), 0700); err != nil { + t.Fatal(err) + } + r, err := Start(context.Background(), dir, time.Now().Add(500*time.Millisecond).Format(time.RFC3339Nano), nil, nil) + if err != nil { + t.Fatal(err) + } + select { + case <-r.done: + case <-time.After(3 * time.Second): + r.Stop() + t.Fatal("did not stop at deadline") + } + profiles, _ := filepath.Glob(filepath.Join(dir, "*.pprof")) + if len(profiles) != 1 { + t.Fatalf("subdirectory stopped the recorder: %v", profiles) + } +} + func TestTriggersAreConfigurable(t *testing.T) { r := &Recorder{triggers: map[string]bool{"custom": true}, requests: make(chan string, 1), done: make(chan struct{})} noop := func(context.Context) error { return nil } From 9ac32fe33c72829f46e0ab61b9ab03b48a2566b4 Mon Sep 17 00:00:00 2001 From: MrAlders0n Date: Wed, 30 Sep 2026 10:28:03 -0400 Subject: [PATCH 07/13] fix(backup): report trailing archive data as corruption --- internal/backup/verify.go | 7 +++++++ internal/backup/verify_test.go | 13 +++++++++++-- 2 files changed, 18 insertions(+), 2 deletions(-) diff --git a/internal/backup/verify.go b/internal/backup/verify.go index f2bf52ae..68847452 100644 --- a/internal/backup/verify.go +++ b/internal/backup/verify.go @@ -53,6 +53,10 @@ func Verify(ctx context.Context, input io.Reader, maxBytes int64) (*Manifest, er // Bound Next itself: it otherwise hides and allocates extension records. decoded.limit = before + padding + 3*512 h, err := tr.Next() + if errors.Is(err, ErrTooLarge) { + // The 3-block header budget is a structure check, not the size limit. + return nil, verificationError(ctx, nil, "tar header") + } if err == io.EOF { // tar.Reader also accepts EOF without both closing zero blocks. if decoded.n-before != padding+2*512 { @@ -106,6 +110,9 @@ func Verify(ctx context.Context, input io.Reader, maxBytes int64) (*Manifest, er decoded.limit = decoded.n var tail [1]byte if _, err := decoded.Read(tail[:]); err != io.EOF { + if errors.Is(err, ErrTooLarge) { + err = nil // bytes after the tar terminator are corruption, not oversize + } return nil, verificationError(ctx, err, "gzip completion or trailing tar data") } if _, err := buffered.ReadByte(); err != io.EOF { diff --git a/internal/backup/verify_test.go b/internal/backup/verify_test.go index fb0e9f1e..06eb1239 100644 --- a/internal/backup/verify_test.go +++ b/internal/backup/verify_test.go @@ -170,9 +170,14 @@ func TestVerifyRejectsInvalidMembersAndManifest(t *testing.T) { } { t.Run(tc.name, func(t *testing.T) { data := verifyGzip(t, verifyTar(t, tc.change(verifyFixture(t)))) - if _, err := Verify(context.Background(), bytes.NewReader(data), int64(len(testSQL))); err == nil || strings.Contains(err.Error(), "PRIVATE_CANARY") { + _, err := Verify(context.Background(), bytes.NewReader(data), int64(len(testSQL))) + if err == nil || strings.Contains(err.Error(), "PRIVATE_CANARY") { t.Fatalf("expected a sanitized rejection, got %v", err) } + // PAX header size errors should be corruption, not size limit. + if tc.name == "hidden oversized metadata" && errors.Is(err, ErrTooLarge) { + t.Fatalf("oversized header reported as size limit: %v", err) + } }) } } @@ -190,9 +195,13 @@ func TestVerifyFraming(t *testing.T) { } for name, data := range cases { t.Run(name, func(t *testing.T) { - if _, err := Verify(context.Background(), bytes.NewReader(data), DefaultMaxBytes); err == nil || strings.Contains(err.Error(), "PRIVATE_CANARY") { + _, err := Verify(context.Background(), bytes.NewReader(data), DefaultMaxBytes) + if err == nil || strings.Contains(err.Error(), "PRIVATE_CANARY") { t.Fatalf("expected sanitized framing error, got %v", err) } + if errors.Is(err, ErrTooLarge) { + t.Fatalf("framing corruption reported as size limit: %v", err) + } }) } } From 71ae82a00786b12ac0021ddaedfe63da76812979 Mon Sep 17 00:00:00 2001 From: MrAlders0n Date: Wed, 30 Sep 2026 10:28:03 -0400 Subject: [PATCH 08/13] perf(db): compute route ambiguity once per reconfirm run The four-way GROUP BY over node_short_ids ran once per 1000-route batch, up to 750 times a run, and could eat the whole 5s batch budget. The task now fetches the set once and passes it to each batch. --- db/queries/queries.sql | 24 +++++++---- db/reconfirm_integration_test.go | 34 +++++++++------ db/routes.go | 26 ++++++++++- db/sqlc/mock/querier.go | 15 +++++++ db/sqlc/querier.go | 5 ++- db/sqlc/queries.sql.go | 62 ++++++++++++++++++++++----- internal/background/reconfirm_test.go | 13 +++++- internal/background/tasks.go | 9 +++- 8 files changed, 152 insertions(+), 36 deletions(-) diff --git a/db/queries/queries.sql b/db/queries/queries.sql index 328960e5..77f1c517 100644 --- a/db/queries/queries.sql +++ b/db/queries/queries.sql @@ -1482,10 +1482,21 @@ REFRESH MATERIALIZED VIEW CONCURRENTLY mv_radio_presets; -- name: RefreshObserverActivity :exec REFRESH MATERIALIZED VIEW CONCURRENTLY mv_observer_activity_hourly; +-- name: AmbiguousPrefixes :many +-- Hop prefixes that match >1 node in an IATA, per width. Computed once per reconfirm run. +SELECT iata::text AS iata, 1::int AS len, prefix_1 AS prefix FROM node_short_ids GROUP BY iata, prefix_1 HAVING COUNT(*) > 1 +UNION ALL +SELECT iata::text, 2::int, prefix_2 FROM node_short_ids GROUP BY iata, prefix_2 HAVING COUNT(*) > 1 +UNION ALL +SELECT iata::text, 3::int, prefix_3 FROM node_short_ids GROUP BY iata, prefix_3 HAVING COUNT(*) > 1 +UNION ALL +SELECT iata::text, 4::int, prefix_4 FROM node_short_ids GROUP BY iata, prefix_4 HAVING COUNT(*) > 1; + -- name: ReconfirmRoutes :one -- Checks one batch of least-recently-reconfirmed routes: deletes those with a departed -- hop node or a hop prefix now matching >1 node in that IATA (length-aware: --- 1/2/3/4-byte hop prefixes check prefix_1/2/3/4), and stamps the survivors. +-- 1/2/3/4-byte hop prefixes check prefix_1/2/3/4; ambiguity set supplied by AmbiguousPrefixes), +-- and stamps the survivors. WITH batch AS MATERIALIZED ( SELECT r.iata, r.path_key, r.node_ids, r.hash_prefix FROM known_routes r @@ -1494,14 +1505,9 @@ WITH batch AS MATERIALIZED ( LIMIT @batch_size FOR UPDATE OF r SKIP LOCKED ), -amb AS MATERIALIZED ( - SELECT iata, 1 AS len, prefix_1 AS p FROM node_short_ids GROUP BY iata, prefix_1 HAVING COUNT(*) > 1 - UNION ALL - SELECT iata, 2, prefix_2 FROM node_short_ids GROUP BY iata, prefix_2 HAVING COUNT(*) > 1 - UNION ALL - SELECT iata, 3, prefix_3 FROM node_short_ids GROUP BY iata, prefix_3 HAVING COUNT(*) > 1 - UNION ALL - SELECT iata, 4, prefix_4 FROM node_short_ids GROUP BY iata, prefix_4 HAVING COUNT(*) > 1 +amb AS ( + SELECT a.iata::char(3) AS iata, a.len, a.p + FROM ROWS FROM (unnest(@amb_iata::text[]), unnest(@amb_len::int[]), unnest(@amb_prefix::bytea[])) AS a(iata, len, p) ), dead AS ( SELECT b.iata, b.path_key diff --git a/db/reconfirm_integration_test.go b/db/reconfirm_integration_test.go index 7aeed34e..0b52206b 100644 --- a/db/reconfirm_integration_test.go +++ b/db/reconfirm_integration_test.go @@ -17,10 +17,19 @@ import ( "github.com/jackc/pgx/v5/pgxpool" ) +func ambiguity(t *testing.T, ctx context.Context, store *Store) AmbiguousPrefixes { + t.Helper() + amb, err := store.AmbiguousPrefixes(ctx) + if err != nil { + t.Fatal(err) + } + return amb +} + func TestReconfirmReleasesEarlierBatchPostgres(t *testing.T) { ctx, pool, store := reconfirmPool(t) before := time.Now() - if n, err := store.ReconfirmRoutes(ctx, 1, before); err != nil || n != 1 { + if n, err := store.ReconfirmRoutes(ctx, 1, before, ambiguity(t, ctx, store)); err != nil || n != 1 { t.Fatalf("first batch: %d, %v", n, err) } gate, err := pool.Begin(ctx) @@ -40,8 +49,9 @@ CREATE TRIGGER pause_second_route BEFORE UPDATE ON known_routes FOR EACH ROW EXE t.Fatal(err) } finished := make(chan error, 1) + amb := ambiguity(t, ctx, store) go func() { - _, err := store.ReconfirmRoutes(ctx, 1, before) + _, err := store.ReconfirmRoutes(ctx, 1, before, amb) finished <- err }() deadline := time.Now().Add(3 * time.Second) @@ -131,7 +141,7 @@ UPDATE known_routes SET node_ids=node_ids || md5('missing')::uuid, hash_prefix=h original := snapshot() var checked int64 for range 10 { - n, err := store.ReconfirmRoutes(ctx, 3, before) + n, err := store.ReconfirmRoutes(ctx, 3, before, ambiguity(t, ctx, store)) if err != nil { t.Fatal(err) } @@ -165,7 +175,7 @@ UPDATE known_routes SET node_ids=node_ids || md5('missing')::uuid, hash_prefix=h func TestReconfirmCancelledBatchRollsBackPostgres(t *testing.T) { ctx, pool, store := reconfirmPool(t) before := time.Now() - if _, err := store.ReconfirmRoutes(ctx, 1, before); err != nil { + if _, err := store.ReconfirmRoutes(ctx, 1, before, ambiguity(t, ctx, store)); err != nil { t.Fatal(err) } _, err := pool.Exec(ctx, ` @@ -177,7 +187,7 @@ CREATE TRIGGER pause_reconfirm BEFORE UPDATE ON known_routes FOR EACH ROW EXECUT } batchCtx, cancel := context.WithTimeout(ctx, 100*time.Millisecond) defer cancel() - if _, err := store.ReconfirmRoutes(batchCtx, 1, before); err == nil { + if _, err := store.ReconfirmRoutes(batchCtx, 1, before, ambiguity(t, ctx, store)); err == nil { t.Fatal("expected batch cancellation") } // Waiting for this DDL also waits for cancellation to release the row lock. @@ -191,7 +201,7 @@ CREATE TRIGGER pause_reconfirm BEFORE UPDATE ON known_routes FOR EACH ROW EXECUT if total != 2 || stamped != 1 { t.Fatalf("cancellation changed committed work: total %d, checked %d", total, stamped) } - if n, err := store.ReconfirmRoutes(ctx, 1, before); err != nil || n != 1 { + if n, err := store.ReconfirmRoutes(ctx, 1, before, ambiguity(t, ctx, store)); err != nil || n != 1 { t.Fatalf("cancelled route was not available to retry: %d, %v", n, err) } } @@ -199,10 +209,10 @@ CREATE TRIGGER pause_reconfirm BEFORE UPDATE ON known_routes FOR EACH ROW EXECUT func TestReconfirmFutureCutoffDoesNotRepeatPostgres(t *testing.T) { ctx, _, store := reconfirmPool(t) before := time.Now().Add(time.Hour) - if n, err := store.ReconfirmRoutes(ctx, 2, before); err != nil || n != 2 { + if n, err := store.ReconfirmRoutes(ctx, 2, before, ambiguity(t, ctx, store)); err != nil || n != 2 { t.Fatalf("first batch: %d, %v", n, err) } - if n, err := store.ReconfirmRoutes(ctx, 2, before); err != nil || n != 0 { + if n, err := store.ReconfirmRoutes(ctx, 2, before, ambiguity(t, ctx, store)); err != nil || n != 0 { t.Fatalf("completed routes consumed the run budget twice: %d, %v", n, err) } } @@ -220,7 +230,7 @@ CREATE TRIGGER pause_valid_route BEFORE UPDATE ON known_routes FOR EACH ROW EXEC before := time.Now() batchCtx, cancel := context.WithTimeout(ctx, 100*time.Millisecond) defer cancel() - if _, err := store.ReconfirmRoutes(batchCtx, 2, before); err == nil { + if _, err := store.ReconfirmRoutes(batchCtx, 2, before, ambiguity(t, ctx, store)); err == nil { t.Fatal("expected batch cancellation") } if _, err := pool.Exec(ctx, `DROP TRIGGER pause_valid_route ON known_routes`); err != nil { @@ -230,7 +240,7 @@ CREATE TRIGGER pause_valid_route BEFORE UPDATE ON known_routes FOR EACH ROW EXEC if err := pool.QueryRow(ctx, `SELECT count(*) FROM known_routes`).Scan(&count); err != nil || count != 2 { t.Fatalf("cancelled batch partially deleted routes: %d, %v", count, err) } - if n, err := store.ReconfirmRoutes(ctx, 2, before); err != nil || n != 2 { + if n, err := store.ReconfirmRoutes(ctx, 2, before, ambiguity(t, ctx, store)); err != nil || n != 2 { t.Fatalf("retry did not process both routes: %d, %v", n, err) } if err := pool.QueryRow(ctx, `SELECT count(*) FROM known_routes WHERE path_key=int4send(2)`).Scan(&count); err != nil || count != 1 { @@ -299,7 +309,7 @@ func TestReconfirmSkipsBusyRoutePostgres(t *testing.T) { } batchCtx, cancel := context.WithTimeout(ctx, 500*time.Millisecond) defer cancel() - if _, err := store.ReconfirmRoutes(batchCtx, 1, time.Now()); err != nil { + if _, err := store.ReconfirmRoutes(batchCtx, 1, time.Now(), ambiguity(t, ctx, store)); err != nil { t.Fatalf("maintenance should skip the busy route and validate the next one: %v", err) } var checked, preserved int @@ -312,7 +322,7 @@ func TestReconfirmSkipsBusyRoutePostgres(t *testing.T) { if err := busy.Rollback(ctx); err != nil { t.Fatal(err) } - if _, err := store.ReconfirmRoutes(ctx, 1, time.Now()); err != nil { + if _, err := store.ReconfirmRoutes(ctx, 1, time.Now(), ambiguity(t, ctx, store)); err != nil { t.Fatal(err) } if err := pool.QueryRow(ctx, `SELECT count(*) FROM known_routes WHERE last_reconfirmed_at > '2026-02-01'`).Scan(&checked); err != nil { diff --git a/db/routes.go b/db/routes.go index 090d2a00..69183fb4 100644 --- a/db/routes.go +++ b/db/routes.go @@ -287,12 +287,36 @@ func (s *Store) SearchCrossIATARoutes(ctx context.Context, fromHash, fromIATA, t return results, nil } +// AmbiguousPrefixes is the per-IATA set of hop prefixes that resolve to more than one node. +type AmbiguousPrefixes struct { + IATAs []string + Lens []int32 + Prefixes [][]byte +} + +func (s *Store) AmbiguousPrefixes(ctx context.Context) (AmbiguousPrefixes, error) { + rows, err := s.q.AmbiguousPrefixes(ctx) + if err != nil { + return AmbiguousPrefixes{}, err + } + amb := AmbiguousPrefixes{IATAs: make([]string, 0, len(rows)), Lens: make([]int32, 0, len(rows)), Prefixes: make([][]byte, 0, len(rows))} + for _, r := range rows { + amb.IATAs = append(amb.IATAs, r.Iata) + amb.Lens = append(amb.Lens, r.Len) + amb.Prefixes = append(amb.Prefixes, r.Prefix) + } + return amb, nil +} + // ReconfirmRoutes checks the batchSize least-recently-reconfirmed routes, // deleting stale or ambiguous ones and stamping the survivors. -func (s *Store) ReconfirmRoutes(ctx context.Context, batchSize int32, before time.Time) (int64, error) { +func (s *Store) ReconfirmRoutes(ctx context.Context, batchSize int32, before time.Time, amb AmbiguousPrefixes) (int64, error) { return s.q.ReconfirmRoutes(ctx, sqlc.ReconfirmRoutesParams{ BatchSize: batchSize, Before: pgtype.Timestamptz{Time: before, Valid: true}, + AmbIata: amb.IATAs, + AmbLen: amb.Lens, + AmbPrefix: amb.Prefixes, }) } diff --git a/db/sqlc/mock/querier.go b/db/sqlc/mock/querier.go index 8a9a7efe..82d16603 100644 --- a/db/sqlc/mock/querier.go +++ b/db/sqlc/mock/querier.go @@ -43,6 +43,21 @@ func (m *MockQuerier) EXPECT() *MockQuerierMockRecorder { return m.recorder } +// AmbiguousPrefixes mocks base method. +func (m *MockQuerier) AmbiguousPrefixes(ctx context.Context) ([]db.AmbiguousPrefixesRow, error) { + m.ctrl.T.Helper() + ret := m.ctrl.Call(m, "AmbiguousPrefixes", ctx) + ret0, _ := ret[0].([]db.AmbiguousPrefixesRow) + ret1, _ := ret[1].(error) + return ret0, ret1 +} + +// AmbiguousPrefixes indicates an expected call of AmbiguousPrefixes. +func (mr *MockQuerierMockRecorder) AmbiguousPrefixes(ctx any) *gomock.Call { + mr.mock.ctrl.T.Helper() + return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "AmbiguousPrefixes", reflect.TypeOf((*MockQuerier)(nil).AmbiguousPrefixes), ctx) +} + // CreateAccount mocks base method. func (m *MockQuerier) CreateAccount(ctx context.Context, name string) (db.Account, error) { m.ctrl.T.Helper() diff --git a/db/sqlc/querier.go b/db/sqlc/querier.go index ef65fd7b..6d699a94 100644 --- a/db/sqlc/querier.go +++ b/db/sqlc/querier.go @@ -12,6 +12,8 @@ import ( ) type Querier interface { + // Hop prefixes that match >1 node in an IATA, per width. Computed once per reconfirm run. + AmbiguousPrefixes(ctx context.Context) ([]AmbiguousPrefixesRow, error) CreateAccount(ctx context.Context, name string) (Account, error) // Lock the current row before deciding the outcome, including when another // deactivation commits while this statement is waiting for its row lock. @@ -223,7 +225,8 @@ type Querier interface { ReconfirmNeighbors(ctx context.Context) error // Checks one batch of least-recently-reconfirmed routes: deletes those with a departed // hop node or a hop prefix now matching >1 node in that IATA (length-aware: - // 1/2/3/4-byte hop prefixes check prefix_1/2/3/4), and stamps the survivors. + // 1/2/3/4-byte hop prefixes check prefix_1/2/3/4; ambiguity set supplied by AmbiguousPrefixes), + // and stamps the survivors. ReconfirmRoutes(ctx context.Context, arg ReconfirmRoutesParams) (int64, error) RefreshHourlyStats(ctx context.Context) error RefreshObserverActivity(ctx context.Context) error diff --git a/db/sqlc/queries.sql.go b/db/sqlc/queries.sql.go index 3b3def70..dcd52c23 100644 --- a/db/sqlc/queries.sql.go +++ b/db/sqlc/queries.sql.go @@ -12,6 +12,43 @@ import ( "github.com/jackc/pgx/v5/pgtype" ) +const ambiguousPrefixes = `-- name: AmbiguousPrefixes :many +SELECT iata::text AS iata, 1::int AS len, prefix_1 AS prefix FROM node_short_ids GROUP BY iata, prefix_1 HAVING COUNT(*) > 1 +UNION ALL +SELECT iata::text, 2::int, prefix_2 FROM node_short_ids GROUP BY iata, prefix_2 HAVING COUNT(*) > 1 +UNION ALL +SELECT iata::text, 3::int, prefix_3 FROM node_short_ids GROUP BY iata, prefix_3 HAVING COUNT(*) > 1 +UNION ALL +SELECT iata::text, 4::int, prefix_4 FROM node_short_ids GROUP BY iata, prefix_4 HAVING COUNT(*) > 1 +` + +type AmbiguousPrefixesRow struct { + Iata string `json:"iata"` + Len int32 `json:"len"` + Prefix []byte `json:"prefix"` +} + +// Hop prefixes that match >1 node in an IATA, per width. Computed once per reconfirm run. +func (q *Queries) AmbiguousPrefixes(ctx context.Context) ([]AmbiguousPrefixesRow, error) { + rows, err := q.db.Query(ctx, ambiguousPrefixes) + if err != nil { + return nil, err + } + defer rows.Close() + items := []AmbiguousPrefixesRow{} + for rows.Next() { + var i AmbiguousPrefixesRow + if err := rows.Scan(&i.Iata, &i.Len, &i.Prefix); err != nil { + return nil, err + } + items = append(items, i) + } + if err := rows.Err(); err != nil { + return nil, err + } + return items, nil +} + const createAccount = `-- name: CreateAccount :one INSERT INTO accounts (name) VALUES ($1) ON CONFLICT (name) WHERE deactivated_at IS NULL DO NOTHING @@ -3972,14 +4009,9 @@ WITH batch AS MATERIALIZED ( LIMIT $2 FOR UPDATE OF r SKIP LOCKED ), -amb AS MATERIALIZED ( - SELECT iata, 1 AS len, prefix_1 AS p FROM node_short_ids GROUP BY iata, prefix_1 HAVING COUNT(*) > 1 - UNION ALL - SELECT iata, 2, prefix_2 FROM node_short_ids GROUP BY iata, prefix_2 HAVING COUNT(*) > 1 - UNION ALL - SELECT iata, 3, prefix_3 FROM node_short_ids GROUP BY iata, prefix_3 HAVING COUNT(*) > 1 - UNION ALL - SELECT iata, 4, prefix_4 FROM node_short_ids GROUP BY iata, prefix_4 HAVING COUNT(*) > 1 +amb AS ( + SELECT a.iata::char(3) AS iata, a.len, a.p + FROM ROWS FROM (unnest($3::text[]), unnest($4::int[]), unnest($5::bytea[])) AS a(iata, len, p) ), dead AS ( SELECT b.iata, b.path_key @@ -4020,13 +4052,23 @@ SELECT count(*) FROM batch type ReconfirmRoutesParams struct { Before pgtype.Timestamptz `json:"before"` BatchSize int32 `json:"batch_size"` + AmbIata []string `json:"amb_iata"` + AmbLen []int32 `json:"amb_len"` + AmbPrefix [][]byte `json:"amb_prefix"` } // Checks one batch of least-recently-reconfirmed routes: deletes those with a departed // hop node or a hop prefix now matching >1 node in that IATA (length-aware: -// 1/2/3/4-byte hop prefixes check prefix_1/2/3/4), and stamps the survivors. +// 1/2/3/4-byte hop prefixes check prefix_1/2/3/4; ambiguity set supplied by AmbiguousPrefixes), +// and stamps the survivors. func (q *Queries) ReconfirmRoutes(ctx context.Context, arg ReconfirmRoutesParams) (int64, error) { - row := q.db.QueryRow(ctx, reconfirmRoutes, arg.Before, arg.BatchSize) + row := q.db.QueryRow(ctx, reconfirmRoutes, + arg.Before, + arg.BatchSize, + arg.AmbIata, + arg.AmbLen, + arg.AmbPrefix, + ) var count int64 err := row.Scan(&count) return count, err diff --git a/internal/background/reconfirm_test.go b/internal/background/reconfirm_test.go index 2012688f..991a7c4c 100644 --- a/internal/background/reconfirm_test.go +++ b/internal/background/reconfirm_test.go @@ -9,10 +9,13 @@ import ( "testing" "testing/synctest" "time" + + "github.com/MeshCore-Beacon/beacon-server/db" ) type reconfirmStore struct { run func(context.Context, int32, time.Time) (int64, error) + ambCalls int neighbors bool } @@ -20,7 +23,12 @@ func (*reconfirmStore) DeleteOldRoutes(context.Context, time.Time, int64, time.T return nil } -func (s *reconfirmStore) ReconfirmRoutes(ctx context.Context, n int32, before time.Time) (int64, error) { +func (s *reconfirmStore) AmbiguousPrefixes(context.Context) (db.AmbiguousPrefixes, error) { + s.ambCalls++ + return db.AmbiguousPrefixes{}, nil +} + +func (s *reconfirmStore) ReconfirmRoutes(ctx context.Context, n int32, before time.Time, _ db.AmbiguousPrefixes) (int64, error) { return s.run(ctx, n, before) } @@ -60,6 +68,9 @@ func TestReconfirmCoverage(t *testing.T) { if left != tc.wantLeft || !s.neighbors { t.Fatalf("remaining %d, neighbor cleanup %t", left, s.neighbors) } + if s.ambCalls != 1 { + t.Fatalf("ambiguity computed %d times, want once per run", s.ambCalls) + } }) } } diff --git a/internal/background/tasks.go b/internal/background/tasks.go index 7d82fdfa..951dca75 100644 --- a/internal/background/tasks.go +++ b/internal/background/tasks.go @@ -95,7 +95,8 @@ const ( type routeMaintainer interface { DeleteOldRoutes(context.Context, time.Time, int64, time.Time) error - ReconfirmRoutes(context.Context, int32, time.Time) (int64, error) + AmbiguousPrefixes(context.Context) (db.AmbiguousPrefixes, error) + ReconfirmRoutes(context.Context, int32, time.Time, db.AmbiguousPrefixes) (int64, error) ReconfirmNeighbors(context.Context) error } @@ -137,10 +138,14 @@ func ReconfirmTask(store routeMaintainer, routeRetention, routeGrace time.Durati if err := store.DeleteOldRoutes(ctx, now.Add(-routeRetention), routeMinObservations, now.Add(-routeGrace)); err != nil { return fmt.Errorf("route retention: %w", err) } + amb, err := store.AmbiguousPrefixes(ctx) + if err != nil { + return fmt.Errorf("ambiguous prefixes: %w", err) + } for remaining := int64(reconfirmRunLimit); remaining > 0; { limit := min(int64(reconfirmBatchSize), remaining) batchCtx, cancel := context.WithTimeout(ctx, reconfirmBatchTimeout) - n, err := store.ReconfirmRoutes(batchCtx, int32(limit), now) + n, err := store.ReconfirmRoutes(batchCtx, int32(limit), now, amb) cancel() if err != nil { return fmt.Errorf("routes: %w", err) From 0a04bba7019cc6f688060de352bb19b795815add Mon Sep 17 00:00:00 2001 From: MrAlders0n Date: Wed, 30 Sep 2026 10:28:03 -0400 Subject: [PATCH 09/13] fix(api): tolerate client clock skew and drop the dead route evidence cursor --- db/route_evidence.go | 2 +- internal/api/handlers/limits.go | 3 +++ internal/api/handlers/observers.go | 2 +- internal/api/handlers/observers_test.go | 9 ++++++++ internal/api/handlers/route_evidence.go | 4 ++-- internal/api/handlers/route_evidence_test.go | 11 +++++++++- internal/api/route_evidence.go | 23 ++++++++++---------- 7 files changed, 38 insertions(+), 16 deletions(-) diff --git a/db/route_evidence.go b/db/route_evidence.go index fc7209e0..e7017ea5 100644 --- a/db/route_evidence.go +++ b/db/route_evidence.go @@ -51,7 +51,7 @@ func (s *Store) GetRouteEvidence(ctx context.Context, iata, key string, query ap } route := toKnownRoutes([]knownRouteRow{{ID: row.ID, NodeIds: row.NodeIds, HashPrefix: row.HashPrefix, Iata: row.Iata, HopCount: row.HopCount, FirstSeen: row.FirstSeen, LastSeen: row.LastSeen, ObservationCount: row.ObservationCount}}, nodes)[0] route.PathKey = hex.EncodeToString(row.PathKey) - out := &api.RouteEvidence{Page: api.Page[api.RouteObservation]{Items: []api.RouteObservation{}}, Route: route, WindowStart: query.Since.UnixMilli(), WindowEnd: query.Until.UnixMilli(), GeneratedAt: time.Now().UnixMilli(), MatchType: "saved_path_prefixes"} + out := &api.RouteEvidence{Items: []api.RouteObservation{}, Route: route, WindowStart: query.Since.UnixMilli(), WindowEnd: query.Until.UnixMilli(), GeneratedAt: time.Now().UnixMilli(), MatchType: "saved_path_prefixes"} width, path, valid := savedRoutePath(row.HashPrefix, row.HopCount) if !valid { return out, nil diff --git a/internal/api/handlers/limits.go b/internal/api/handlers/limits.go index dfc48233..80bda95b 100644 --- a/internal/api/handlers/limits.go +++ b/internal/api/handlers/limits.go @@ -8,10 +8,13 @@ import ( "log/slog" "net/http" "strconv" + "time" ) const maxListLimit int32 = 200 +const clockSkewTolerance = 5 * time.Minute // client clocks run ahead; accept a small future 'until' + // parseLimit caps the requested result size while retaining each endpoint's default. func parseLimit(r *http.Request, defaultLimit int32) (int32, error) { value := r.URL.Query().Get("limit") diff --git a/internal/api/handlers/observers.go b/internal/api/handlers/observers.go index 66bddd8d..89fddd68 100644 --- a/internal/api/handlers/observers.go +++ b/internal/api/handlers/observers.go @@ -300,7 +300,7 @@ func getObserverActivity(reader api.Reader) http.HandlerFunc { if values, ok := r.URL.Query()["until"]; ok { n, err := strconv.ParseInt(values[0], 10, 64) now := time.Now() - if len(values) != 1 || err != nil || n < 0 || n > now.UnixMilli() || n < now.Add(-720*time.Hour).UnixMilli() { + if len(values) != 1 || err != nil || n < 0 || n > now.Add(clockSkewTolerance).UnixMilli() || n < now.Add(-720*time.Hour).UnixMilli() { respondError(w, http.StatusBadRequest, "until must be one past epoch-millisecond timestamp within 30 days") return } diff --git a/internal/api/handlers/observers_test.go b/internal/api/handlers/observers_test.go index cfef1c9c..7344526e 100644 --- a/internal/api/handlers/observers_test.go +++ b/internal/api/handlers/observers_test.go @@ -7,6 +7,7 @@ import ( "context" "encoding/json" "errors" + "fmt" "net/http" "net/http/httptest" "testing" @@ -225,6 +226,14 @@ func TestGetObserverActivity_Defaults(t *testing.T) { if body.Range != "24h" || body.Interval != "15m" { t.Errorf("expected range 24h interval 15m, got %q/%q", body.Range, body.Interval) } + + skewed := time.Now().Add(time.Minute).UnixMilli() + req = httptest.NewRequest(http.MethodGet, fmt.Sprintf("/observers/%s/activity?until=%d", observerID.String(), skewed), nil) + w = httptest.NewRecorder() + r.ServeHTTP(w, req) + if w.Code != http.StatusOK { + t.Fatalf("expected 200 for until within clock skew, got %d (%s)", w.Code, w.Body.String()) + } } func TestGetObserverActivity_CustomRangeAndInterval(t *testing.T) { diff --git a/internal/api/handlers/route_evidence.go b/internal/api/handlers/route_evidence.go index 6c156fef..a194306c 100644 --- a/internal/api/handlers/route_evidence.go +++ b/internal/api/handlers/route_evidence.go @@ -67,7 +67,7 @@ func routeEvidenceQuery(r *http.Request, iata, key string, now time.Time) (api.R return q, api.ErrRouteEvidenceInput } } - if !api.ValidRouteEvidenceWindow(q.Since, q.Until) || q.Until.After(now) { + if !api.ValidRouteEvidenceWindow(q.Since, q.Until) || q.Until.After(now.Add(clockSkewTolerance)) { return q, api.ErrRouteEvidenceInput } return q, nil @@ -83,7 +83,7 @@ func routeEvidenceQuery(r *http.Request, iata, key string, now time.Time) (api.R // @Param pathKey path string true "Stable 32-hex pathKey from a known-route response" // @Param range query string false "Server-anchored duration (default 24h, max 720h); exclusive with since/until/pageCursor" // @Param since query int false "Window start epoch ms; provide with until (default last 24h)" -// @Param until query int false "Exclusive window end epoch ms; maximum span 30d, no future end" +// @Param until query int false "Exclusive window end epoch ms; maximum span 30d, end may be up to 5 minutes ahead of server time (clock skew tolerance)" // @Param pageCursor query string false "Opaque precise cursor from nextPageCursor; window is pinned" // @Param limit query int false "Default 50; positive, capped at 200" minimum(1) maximum(200) // @Success 200 {object} api.RouteEvidence diff --git a/internal/api/handlers/route_evidence_test.go b/internal/api/handlers/route_evidence_test.go index 8340d85b..f2268d86 100644 --- a/internal/api/handlers/route_evidence_test.go +++ b/internal/api/handlers/route_evidence_test.go @@ -5,6 +5,7 @@ package handlers import ( "context" "errors" + "fmt" "net/http/httptest" "net/url" "testing" @@ -44,6 +45,14 @@ func TestRouteEvidenceQuery(t *testing.T) { t.Errorf("accepted %s", query) } } + ahead := now.Add(time.Minute) + if _, err := routeEvidenceQuery(httptest.NewRequest("GET", fmt.Sprintf("/?since=%d&until=%d", ahead.Add(-time.Hour).UnixMilli(), ahead.UnixMilli()), nil), "YOW", evidenceKey, now); err != nil { + t.Fatalf("rejected until within clock skew: %v", err) + } + far := now.Add(10 * time.Minute) + if _, err := routeEvidenceQuery(httptest.NewRequest("GET", fmt.Sprintf("/?since=%d&until=%d", far.Add(-time.Hour).UnixMilli(), far.UnixMilli()), nil), "YOW", evidenceKey, now); err == nil { + t.Fatal("accepted until far in the future") + } } func TestRouteEvidenceHandler(t *testing.T) { @@ -64,7 +73,7 @@ func TestRouteEvidenceHandler(t *testing.T) { if iata != "YOW" || key != evidenceKey || q.Limit != 50 { t.Fatal("wrong reader scope") } - return &api.RouteEvidence{Page: api.Page[api.RouteObservation]{Items: []api.RouteObservation{}}}, tc.err + return &api.RouteEvidence{Items: []api.RouteObservation{}}, tc.err }}) w := httptest.NewRecorder() router.ServeHTTP(w, httptest.NewRequest("GET", tc.path, nil)) diff --git a/internal/api/route_evidence.go b/internal/api/route_evidence.go index cadc056a..d320ade5 100644 --- a/internal/api/route_evidence.go +++ b/internal/api/route_evidence.go @@ -30,18 +30,19 @@ type RouteObservation struct { SNR *float32 `json:"snr,omitempty"` } -// RouteEvidence uses nextPageCursor rather than the lossy legacy numeric cursor. +// RouteEvidence pages by nextPageCursor, which keeps full timestamp precision. type RouteEvidence struct { - Page[RouteObservation] - Route KnownRoute `json:"route"` - WindowStart int64 `json:"windowStart"` - WindowEnd int64 `json:"windowEnd"` - GeneratedAt int64 `json:"generatedAt"` - MatchType string `json:"matchType"` - MatchAvailable bool `json:"matchAvailable"` - HashSize int16 `json:"hashSize"` - PathBytes string `json:"pathBytes"` - NextPageCursor *string `json:"nextPageCursor,omitempty"` + Items []RouteObservation `json:"items"` + HasMore bool `json:"hasMore"` + Route KnownRoute `json:"route"` + WindowStart int64 `json:"windowStart"` + WindowEnd int64 `json:"windowEnd"` + GeneratedAt int64 `json:"generatedAt"` + MatchType string `json:"matchType"` + MatchAvailable bool `json:"matchAvailable"` + HashSize int16 `json:"hashSize"` + PathBytes string `json:"pathBytes"` + NextPageCursor *string `json:"nextPageCursor,omitempty"` } type RouteEvidenceQuery struct { From 326a807a6de75dde4f3e92dcf566595f9a2bddc1 Mon Sep 17 00:00:00 2001 From: MrAlders0n Date: Wed, 30 Sep 2026 10:28:03 -0400 Subject: [PATCH 10/13] fix(cache): invalidate derived scope keys on catalogue import --- internal/cache/cache.go | 18 ++++++++++++++++++ internal/cache/cache_test.go | 13 +++++++++++++ internal/cache/reader.go | 4 +++- 3 files changed, 34 insertions(+), 1 deletion(-) diff --git a/internal/cache/cache.go b/internal/cache/cache.go index 92377544..48fb5694 100644 --- a/internal/cache/cache.go +++ b/internal/cache/cache.go @@ -76,6 +76,24 @@ func (c *Client) del(ctx context.Context, keys ...string) { c.rdb.Del(ctx, keys...) } +// delPrefix SCANs the whole keyspace; call it rarely. +func (c *Client) delPrefix(ctx context.Context, prefix string) { + var cursor uint64 + for { + keys, next, err := c.rdb.Scan(ctx, cursor, prefix+"*", 100).Result() + if err != nil { + return + } + if len(keys) > 0 { + c.rdb.Del(ctx, keys...) + } + if next == 0 { + return + } + cursor = next + } +} + // getOrSet retrieves a cached value by key, deserialising it into T. // On a cache miss it calls fetch, stores the result under key with the given // TTL, and returns it. If Redis is unavailable or returns an unexpected error, diff --git a/internal/cache/cache_test.go b/internal/cache/cache_test.go index 8184a17b..22515312 100644 --- a/internal/cache/cache_test.go +++ b/internal/cache/cache_test.go @@ -383,6 +383,19 @@ func TestCachedReader_IATASortingForStableKey(t *testing.T) { } } +func TestInvalidateScopeNamesClearsDerivedKeys(t *testing.T) { + c, mr := newTestClient(t) + cr := &CachedReader{inner: &stubReader{}, c: c, ttl: CacheTTLs{}} + for _, k := range []string{keyScopeNames, keyScopesByIATAsPrefix + "YOW", keyScopesByIATAsPrefix + "YOW,YVR", keyScopeByNamePrefix + "#yow"} { + mr.Set(k, "x") + } + mr.Set("beacon:iatas", "keep") + cr.InvalidateScopeNames(context.Background()) + if got := mr.Keys(); len(got) != 1 || got[0] != "beacon:iatas" { + t.Fatalf("keys left: %v", got) + } +} + func TestCachedReader_InvalidateNode(t *testing.T) { c, mr := newTestClient(t) nodeID := uuid.MustParse("00000000-0000-0000-0000-000000000001") diff --git a/internal/cache/reader.go b/internal/cache/reader.go index 515adc93..2dd3dda2 100644 --- a/internal/cache/reader.go +++ b/internal/cache/reader.go @@ -80,9 +80,11 @@ func NewCachedReader(inner api.Reader, c *Client, ttl CacheTTLs) api.Reader { } } -// InvalidateScopeNames makes newly committed catalogue names available to filters. +// InvalidateScopeNames makes newly committed catalogue names visible to every scope read. func (cr *CachedReader) InvalidateScopeNames(ctx context.Context) { cr.c.del(ctx, keyScopeNames) + cr.c.delPrefix(ctx, keyScopesByIATAsPrefix) + cr.c.delPrefix(ctx, keyScopeByNamePrefix) } // InvalidateNode removes the cached entries for a node by UUID. From 916e1260fc0242f4a519685bc324d4d4c7b0ea24 Mon Sep 17 00:00:00 2001 From: MrAlders0n Date: Wed, 30 Sep 2026 10:28:03 -0400 Subject: [PATCH 11/13] fix(meshmapper): retry failed catalogue fetches sooner A transient failure pushed the next attempt out by the full refresh interval. Failures now retry in five minutes, with a server Retry-After still honoured as a ceiling. --- internal/meshmapper/scopes.go | 15 +++++++++--- internal/meshmapper/scopes_test.go | 39 ++++++++++++++++++++++++++++++ 2 files changed, 50 insertions(+), 4 deletions(-) diff --git a/internal/meshmapper/scopes.go b/internal/meshmapper/scopes.go index 23363e69..5eb0d7bb 100644 --- a/internal/meshmapper/scopes.go +++ b/internal/meshmapper/scopes.go @@ -26,6 +26,8 @@ const ( MaxScopes = 64 // per source; at most 16 configured sources (1024 imported identities) MaxBody = 64 << 10 PollInterval = 15 * time.Second // one source per tick, at most four requests/minute + + failureRetry = 5 * time.Minute // transient failures shouldn't cost a whole refresh interval ) // Cache stores source provenance and freshness separately from packet evidence. @@ -142,6 +144,7 @@ func (i *Importer) refresh(ctx context.Context, s *source, now time.Time) error response, err := i.client.Do(request) var entries []scopestore.Entry var generated time.Time + var retryAfter time.Time if err != nil { if ctx.Err() != nil { return ctx.Err() @@ -169,10 +172,7 @@ func (i *Importer) refresh(ctx context.Context, s *source, now time.Time) error if response.StatusCode == http.StatusTooManyRequests || response.StatusCode == http.StatusServiceUnavailable { // The published contract uses seconds. Ignore invalid/overflowing headers. if seconds, parseErr := strconv.ParseInt(response.Header.Get("Retry-After"), 10, 64); parseErr == nil && seconds > 0 && seconds <= int64((1<<63-1)/time.Second) { - until := now.Add(time.Duration(seconds) * time.Second) - if until.After(update.NextAttempt) { - update.NextAttempt = until - } + retryAfter = now.Add(time.Duration(seconds) * time.Second) } } } @@ -191,6 +191,13 @@ func (i *Importer) refresh(ctx context.Context, s *source, now time.Time) error } } } + if update.LastError != "" { + // Transient failures shouldn't cost a whole refresh interval, but never beat a Retry-After. + update.NextAttempt = now.Add(min(i.interval, failureRetry)) + if retryAfter.After(update.NextAttempt) { + update.NextAttempt = retryAfter + } + } if update.LastError == "HTTP 429" { i.retryAfter = update.NextAttempt } diff --git a/internal/meshmapper/scopes_test.go b/internal/meshmapper/scopes_test.go index 858ea582..b8e1b24d 100644 --- a/internal/meshmapper/scopes_test.go +++ b/internal/meshmapper/scopes_test.go @@ -278,3 +278,42 @@ func TestInvalidSavedCatalogueDoesNotBlockStartupOrRefresh(t *testing.T) { t.Fatalf("normal shutdown is a task failure: %v", err) } } + +func TestRetryAfterShorterThanIntervalIsHonoured(t *testing.T) { + ctx := context.Background() + server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) { + w.Header().Set("Retry-After", "1800") + w.WriteHeader(http.StatusTooManyRequests) + })) + defer server.Close() + cfg := config.MeshMapperScopesConfig{Enabled: true, Sources: map[string]string{"YOW": server.URL}} + imp, err := New(ctx, cfg, &memoryStore{rows: map[string]Cache{}}, scopestore.New(), nil) + if err != nil { + t.Fatal(err) + } + now := time.Now().UTC() + if err := imp.refresh(ctx, &imp.sources[0], now); err != nil { + t.Fatal(err) + } + if got := imp.sources[0].cache.NextAttempt.Sub(now); got != 30*time.Minute { + t.Fatalf("next attempt after 429 in %v, want %v (server Retry-After must not be overridden by the 1h interval)", got, 30*time.Minute) + } +} + +func TestFailedRefreshRetriesSoon(t *testing.T) { + ctx := context.Background() + server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) { w.WriteHeader(500) })) + defer server.Close() + cfg := config.MeshMapperScopesConfig{Enabled: true, Sources: map[string]string{"YOW": server.URL}} + imp, err := New(ctx, cfg, &memoryStore{rows: map[string]Cache{}}, scopestore.New(), nil) + if err != nil { + t.Fatal(err) + } + now := time.Now().UTC() + if err := imp.refresh(ctx, &imp.sources[0], now); err != nil { + t.Fatal(err) + } + if got := imp.sources[0].cache.NextAttempt.Sub(now); got != failureRetry { + t.Fatalf("next attempt after failure in %v, want %v", got, failureRetry) + } +} From d77e9cbaa96ad4bf619ad4a60094b49c70c9bac3 Mon Sep 17 00:00:00 2001 From: MrAlders0n Date: Wed, 30 Sep 2026 10:28:03 -0400 Subject: [PATCH 12/13] fix(api): emit observer statusMetadata as JSON --- db/observers.go | 3 ++- db/observers_test.go | 12 ++++++++++++ internal/api/observers.go | 18 +++++++++++------- 3 files changed, 25 insertions(+), 8 deletions(-) diff --git a/db/observers.go b/db/observers.go index fc10efad..f312aca1 100644 --- a/db/observers.go +++ b/db/observers.go @@ -6,6 +6,7 @@ package db import ( "context" "encoding/hex" + "encoding/json" "fmt" "log/slog" "math" @@ -116,7 +117,7 @@ func (s *Store) GetObserver(ctx context.Context, observerID uuid.UUID) (*api.Obs RadioCR: obs.RadioCr, BatteryLevel: obs.BatteryLevel, UptimeSeconds: obs.UptimeSeconds, - StatusMetadata: obs.StatusMetadata, + StatusMetadata: json.RawMessage(obs.StatusMetadata), FirstSeen: obs.FirstSeen.Time.UnixMilli(), LastSeen: obs.LastSeen.Time.UnixMilli(), ObservationCount: *obs.ObservationCount, diff --git a/db/observers_test.go b/db/observers_test.go index f6c67d3d..e5b98506 100644 --- a/db/observers_test.go +++ b/db/observers_test.go @@ -5,7 +5,9 @@ package db import ( "context" + "encoding/json" "errors" + "strings" "testing" "time" @@ -178,6 +180,7 @@ func TestGetObserver_OnlineStatus(t *testing.T) { observerID := uuid.MustParse("00000000-0000-0000-0000-000000000001") obsCount := int64(10) + statusMetadata := []byte(`{"stats":{"noise_floor":-97}}`) mock.EXPECT(). GetObserverByID(gomock.Any(), observerID). @@ -188,6 +191,7 @@ func TestGetObserver_OnlineStatus(t *testing.T) { FirstSeen: pgtype.Timestamptz{Time: time.Now().Add(-time.Hour), Valid: true}, LastSeen: pgtype.Timestamptz{Time: time.Now().Add(-time.Minute), Valid: true}, LastStatusAt: pgtype.Timestamptz{Time: time.Now().Add(-time.Minute), Valid: true}, + StatusMetadata: statusMetadata, }, nil) mock.EXPECT(). @@ -213,6 +217,14 @@ func TestGetObserver_OnlineStatus(t *testing.T) { if observer.IATA != "YVR" { t.Errorf("expected IATA YVR, got %s", observer.IATA) } + + out, err := json.Marshal(observer) + if err != nil { + t.Fatal(err) + } + if !strings.Contains(string(out), `"statusMetadata":{"stats":{"noise_floor":-97}}`) { + t.Fatalf("statusMetadata not an object: %s", out) + } } func TestGetObserver_OfflineStatus(t *testing.T) { diff --git a/internal/api/observers.go b/internal/api/observers.go index 2da680c1..080b74d7 100644 --- a/internal/api/observers.go +++ b/internal/api/observers.go @@ -3,7 +3,11 @@ package api -import "github.com/google/uuid" +import ( + "encoding/json" + + "github.com/google/uuid" +) // ObserverSummary is the minimal observer representation used in list responses. type ObserverSummary struct { @@ -40,12 +44,12 @@ type Observer struct { RadioCR *int16 `json:"radioCr,omitempty"` // coding rate denominator BatteryLevel *float32 `json:"batteryLevel,omitempty"` // volts, nil if mains powered UptimeSeconds *int64 `json:"uptimeSeconds,omitempty"` - StatusMetadata any `json:"statusMetadata,omitempty"` // raw /status JSON payload - LastStatusAt *int64 `json:"lastStatusAt,omitempty"` // epoch ms - FirstSeen int64 `json:"firstSeen"` // epoch ms - LastSeen int64 `json:"lastSeen"` // epoch ms - ObservationCount int64 `json:"observationCount"` // legacy cumulative presence counter; includes non-packet events - Brokers []ObserverBroker `json:"brokers"` // broker names this observer has been seen on + StatusMetadata json.RawMessage `json:"statusMetadata,omitempty" swaggertype:"object"` // raw /status JSON payload + LastStatusAt *int64 `json:"lastStatusAt,omitempty"` // epoch ms + FirstSeen int64 `json:"firstSeen"` // epoch ms + LastSeen int64 `json:"lastSeen"` // epoch ms + ObservationCount int64 `json:"observationCount"` // legacy cumulative presence counter; includes non-packet events + Brokers []ObserverBroker `json:"brokers"` // broker names this observer has been seen on } // ObserverTelemetryPoint is a single telemetry snapshot for an observer. From 4be63ece9938ec3474553a94c97b26a39c0b8085 Mon Sep 17 00:00:00 2001 From: MrAlders0n Date: Wed, 30 Sep 2026 10:28:03 -0400 Subject: [PATCH 13/13] docs: regenerate swagger --- docs/docs.go | 10 ++++------ docs/swagger.json | 10 ++++------ docs/swagger.yaml | 8 ++++---- 3 files changed, 12 insertions(+), 16 deletions(-) diff --git a/docs/docs.go b/docs/docs.go index 288a70b6..df1ccc9e 100644 --- a/docs/docs.go +++ b/docs/docs.go @@ -2170,7 +2170,7 @@ const docTemplate = `{ }, { "type": "integer", - "description": "Exclusive window end epoch ms; maximum span 30d, no future end", + "description": "Exclusive window end epoch ms; maximum span 30d, end may be up to 5 minutes ahead of server time (clock skew tolerance)", "name": "until", "in": "query" }, @@ -4080,7 +4080,8 @@ const docTemplate = `{ "type": "string" }, "statusMetadata": { - "description": "raw /status JSON payload" + "description": "raw /status JSON payload", + "type": "object" }, "uptimeSeconds": { "type": "integer" @@ -4631,7 +4632,7 @@ const docTemplate = `{ "type": "string" }, "summary": { - "description": "advert name from this packet; omitted when unavailable or unsupported", + "description": "advert name, or the ACK/TRACE/PING summary; omitted when unavailable", "type": "string" } } @@ -4973,9 +4974,6 @@ const docTemplate = `{ "matchType": { "type": "string" }, - "nextCursor": { - "type": "integer" - }, "nextPageCursor": { "type": "string" }, diff --git a/docs/swagger.json b/docs/swagger.json index 56794446..183611af 100644 --- a/docs/swagger.json +++ b/docs/swagger.json @@ -2168,7 +2168,7 @@ }, { "type": "integer", - "description": "Exclusive window end epoch ms; maximum span 30d, no future end", + "description": "Exclusive window end epoch ms; maximum span 30d, end may be up to 5 minutes ahead of server time (clock skew tolerance)", "name": "until", "in": "query" }, @@ -4078,7 +4078,8 @@ "type": "string" }, "statusMetadata": { - "description": "raw /status JSON payload" + "description": "raw /status JSON payload", + "type": "object" }, "uptimeSeconds": { "type": "integer" @@ -4629,7 +4630,7 @@ "type": "string" }, "summary": { - "description": "advert name from this packet; omitted when unavailable or unsupported", + "description": "advert name, or the ACK/TRACE/PING summary; omitted when unavailable", "type": "string" } } @@ -4971,9 +4972,6 @@ "matchType": { "type": "string" }, - "nextCursor": { - "type": "integer" - }, "nextPageCursor": { "type": "string" }, diff --git a/docs/swagger.yaml b/docs/swagger.yaml index a58626f2..ca79d350 100644 --- a/docs/swagger.yaml +++ b/docs/swagger.yaml @@ -570,6 +570,7 @@ definitions: type: string statusMetadata: description: raw /status JSON payload + type: object uptimeSeconds: type: integer type: object @@ -956,7 +957,7 @@ definitions: description: matched transport scope name e.g. "#bc" type: string summary: - description: advert name from this packet; omitted when unavailable or unsupported + description: advert name, or the ACK/TRACE/PING summary; omitted when unavailable type: string type: object github_com_MeshCore-Beacon_beacon-server_internal_api.PacketTransportCodes: @@ -1183,8 +1184,6 @@ definitions: type: boolean matchType: type: string - nextCursor: - type: integer nextPageCursor: type: string pathBytes: @@ -2894,7 +2893,8 @@ paths: in: query name: since type: integer - - description: Exclusive window end epoch ms; maximum span 30d, no future end + - description: Exclusive window end epoch ms; maximum span 30d, end may be up + to 5 minutes ahead of server time (clock skew tolerance) in: query name: until type: integer