diff --git a/cmd/server/backfill_async_test.go b/cmd/server/backfill_async_test.go new file mode 100644 index 00000000..b3df8446 --- /dev/null +++ b/cmd/server/backfill_async_test.go @@ -0,0 +1,132 @@ +package main + +import ( + "encoding/json" + "net/http" + "net/http/httptest" + "testing" + "time" + + "github.com/gorilla/mux" +) + +// TestBackfillAsyncChunked verifies that backfillResolvedPathsAsync processes +// observations in chunks, yields between batches, and sets the completion flag. +func TestBackfillAsyncChunked(t *testing.T) { + store := &PacketStore{ + packets: make([]*StoreTx, 0), + byHash: make(map[string]*StoreTx), + byTxID: make(map[int]*StoreTx), + byObsID: make(map[int]*StoreObs), + } + + // No pending observations → should complete immediately. + backfillResolvedPathsAsync(store, "", 100, time.Millisecond, 24) + if !store.backfillComplete.Load() { + t.Fatal("expected backfillComplete to be true with empty store") + } +} + +// TestBackfillStatusHeader verifies the X-CoreScope-Status header is set correctly. +func TestBackfillStatusHeader(t *testing.T) { + store := &PacketStore{ + packets: make([]*StoreTx, 0), + byHash: make(map[string]*StoreTx), + byTxID: make(map[int]*StoreTx), + byObsID: make(map[int]*StoreObs), + } + + srv := &Server{store: store} + + handler := srv.backfillStatusMiddleware(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + w.WriteHeader(200) + })) + + // Before backfill completes → backfilling + req := httptest.NewRequest("GET", "/api/stats", nil) + rec := httptest.NewRecorder() + handler.ServeHTTP(rec, req) + if got := rec.Header().Get("X-CoreScope-Status"); got != "backfilling" { + t.Fatalf("expected 'backfilling', got %q", got) + } + + // After backfill completes → ready + store.backfillComplete.Store(true) + rec = httptest.NewRecorder() + handler.ServeHTTP(rec, req) + if got := rec.Header().Get("X-CoreScope-Status"); got != "ready" { + t.Fatalf("expected 'ready', got %q", got) + } +} + +// TestStatsBackfillFields verifies /api/stats includes backfill fields. +func TestStatsBackfillFields(t *testing.T) { + db := setupTestDBv2(t) + defer db.Close() + seedV2Data(t, db) + + store := &PacketStore{ + db: db, + packets: make([]*StoreTx, 0), + byHash: make(map[string]*StoreTx), + byTxID: make(map[int]*StoreTx), + byObsID: make(map[int]*StoreObs), + loaded: true, + } + + cfg := &Config{Port: 0} + hub := NewHub() + srv := NewServer(db, cfg, hub) + srv.store = store + + router := mux.NewRouter() + srv.RegisterRoutes(router) + + // While backfilling + req := httptest.NewRequest("GET", "/api/stats", nil) + rec := httptest.NewRecorder() + router.ServeHTTP(rec, req) + + var resp map[string]interface{} + if err := json.Unmarshal(rec.Body.Bytes(), &resp); err != nil { + t.Fatalf("failed to parse stats response: %v", err) + } + + if backfilling, ok := resp["backfilling"]; !ok { + t.Fatal("missing 'backfilling' field in stats response") + } else if backfilling != true { + t.Fatalf("expected backfilling=true, got %v", backfilling) + } + + if _, ok := resp["backfillProgress"]; !ok { + t.Fatal("missing 'backfillProgress' field in stats response") + } + + // Check header + if got := rec.Header().Get("X-CoreScope-Status"); got != "backfilling" { + t.Fatalf("expected X-CoreScope-Status=backfilling, got %q", got) + } + + // After backfill completes + store.backfillComplete.Store(true) + // Invalidate stats cache + srv.statsMu.Lock() + srv.statsCache = nil + srv.statsMu.Unlock() + + rec = httptest.NewRecorder() + router.ServeHTTP(rec, req) + + resp = nil + if err := json.Unmarshal(rec.Body.Bytes(), &resp); err != nil { + t.Fatalf("failed to parse stats response: %v", err) + } + + if backfilling, ok := resp["backfilling"]; !ok || backfilling != false { + t.Fatalf("expected backfilling=false after completion, got %v", backfilling) + } + + if got := rec.Header().Get("X-CoreScope-Status"); got != "ready" { + t.Fatalf("expected X-CoreScope-Status=ready, got %q", got) + } +} diff --git a/cmd/server/config.go b/cmd/server/config.go index 442a996f..9bd40abc 100644 --- a/cmd/server/config.go +++ b/cmd/server/config.go @@ -57,6 +57,19 @@ type Config struct { Timestamps *TimestampConfig `json:"timestamps,omitempty"` DebugAffinity bool `json:"debugAffinity,omitempty"` + + ResolvedPath *ResolvedPathConfig `json:"resolvedPath,omitempty"` + NeighborGraph *NeighborGraphConfig `json:"neighborGraph,omitempty"` +} + +// ResolvedPathConfig controls async backfill behavior. +type ResolvedPathConfig struct { + BackfillHours int `json:"backfillHours"` // how far back (hours) to scan for NULL resolved_path (default 24) +} + +// NeighborGraphConfig controls neighbor edge pruning. +type NeighborGraphConfig struct { + MaxAgeDays int `json:"maxAgeDays"` // edges older than this are pruned (default 5) } // PacketStoreConfig controls in-memory packet store limits. @@ -82,6 +95,21 @@ func (c *Config) MetricsRetentionDays() int { return 30 } +// BackfillHours returns configured backfill window or 24h default. +func (c *Config) BackfillHours() int { + if c.ResolvedPath != nil && c.ResolvedPath.BackfillHours > 0 { + return c.ResolvedPath.BackfillHours + } + return 24 +} + +// NeighborMaxAgeDays returns configured max edge age or 30 days default. +func (c *Config) NeighborMaxAgeDays() int { + if c.NeighborGraph != nil && c.NeighborGraph.MaxAgeDays > 0 { + return c.NeighborGraph.MaxAgeDays + } + return 5 +} type TimestampConfig struct { DefaultMode string `json:"defaultMode"` // "ago" | "absolute" diff --git a/cmd/server/config_knobs_test.go b/cmd/server/config_knobs_test.go new file mode 100644 index 00000000..9a3cd4fa --- /dev/null +++ b/cmd/server/config_knobs_test.go @@ -0,0 +1,177 @@ +package main + +import ( + "database/sql" + "path/filepath" + "testing" + "time" + + _ "modernc.org/sqlite" +) + +func TestBackfillHoursDefault(t *testing.T) { + cfg := &Config{} + if got := cfg.BackfillHours(); got != 24 { + t.Errorf("BackfillHours() = %d, want 24", got) + } +} + +func TestBackfillHoursConfigured(t *testing.T) { + cfg := &Config{ResolvedPath: &ResolvedPathConfig{BackfillHours: 48}} + if got := cfg.BackfillHours(); got != 48 { + t.Errorf("BackfillHours() = %d, want 48", got) + } +} + +func TestBackfillHoursZeroFallsBack(t *testing.T) { + cfg := &Config{ResolvedPath: &ResolvedPathConfig{BackfillHours: 0}} + if got := cfg.BackfillHours(); got != 24 { + t.Errorf("BackfillHours() = %d, want 24 (default for zero)", got) + } +} + +func TestNeighborMaxAgeDaysDefault(t *testing.T) { + cfg := &Config{} + if got := cfg.NeighborMaxAgeDays(); got != 5 { + t.Errorf("NeighborMaxAgeDays() = %d, want 5", got) + } +} + +func TestNeighborMaxAgeDaysConfigured(t *testing.T) { + cfg := &Config{NeighborGraph: &NeighborGraphConfig{MaxAgeDays: 7}} + if got := cfg.NeighborMaxAgeDays(); got != 7 { + t.Errorf("NeighborMaxAgeDays() = %d, want 7", got) + } +} + +func TestGraphPruneOlderThan(t *testing.T) { + g := NewNeighborGraph() + now := time.Now().UTC() + + // Add a recent edge + g.upsertEdge("aaa", "bbb", "bb", "obs1", nil, now) + // Add an old edge + g.upsertEdge("ccc", "ddd", "dd", "obs1", nil, now.Add(-60*24*time.Hour)) + + if len(g.AllEdges()) != 2 { + t.Fatalf("expected 2 edges, got %d", len(g.AllEdges())) + } + + cutoff := now.Add(-30 * 24 * time.Hour) + pruned := g.PruneOlderThan(cutoff) + if pruned != 1 { + t.Errorf("PruneOlderThan pruned %d, want 1", pruned) + } + + edges := g.AllEdges() + if len(edges) != 1 { + t.Fatalf("expected 1 edge after prune, got %d", len(edges)) + } + if edges[0].NodeA != "aaa" && edges[0].NodeB != "aaa" { + t.Errorf("wrong edge survived prune: %+v", edges[0]) + } +} + +func TestPruneNeighborEdgesDB(t *testing.T) { + tmpDir := t.TempDir() + dbPath := filepath.Join(tmpDir, "test.db") + db, err := sql.Open("sqlite", "file:"+dbPath+"?_journal_mode=WAL") + if err != nil { + t.Fatal(err) + } + defer db.Close() + + _, err = db.Exec(`CREATE TABLE neighbor_edges ( + node_a TEXT NOT NULL, + node_b TEXT NOT NULL, + count INTEGER DEFAULT 1, + last_seen TEXT, + PRIMARY KEY (node_a, node_b) + )`) + if err != nil { + t.Fatal(err) + } + + now := time.Now().UTC() + old := now.Add(-60 * 24 * time.Hour) + + db.Exec("INSERT INTO neighbor_edges (node_a, node_b, count, last_seen) VALUES (?, ?, 5, ?)", + "aaa", "bbb", now.Format(time.RFC3339)) + db.Exec("INSERT INTO neighbor_edges (node_a, node_b, count, last_seen) VALUES (?, ?, 3, ?)", + "ccc", "ddd", old.Format(time.RFC3339)) + + g := NewNeighborGraph() + g.upsertEdge("aaa", "bbb", "bb", "obs1", nil, now) + g.upsertEdge("ccc", "ddd", "dd", "obs1", nil, old) + + pruned, err := PruneNeighborEdges(dbPath, g, 30) + if err != nil { + t.Fatal(err) + } + if pruned != 1 { + t.Errorf("PruneNeighborEdges pruned %d DB rows, want 1", pruned) + } + + var count int + db.QueryRow("SELECT COUNT(*) FROM neighbor_edges").Scan(&count) + if count != 1 { + t.Errorf("expected 1 row in DB after prune, got %d", count) + } + + if len(g.AllEdges()) != 1 { + t.Errorf("expected 1 in-memory edge after prune, got %d", len(g.AllEdges())) + } +} + +func TestBackfillRespectsHourWindow(t *testing.T) { + store := &PacketStore{} + + now := time.Now().UTC() + oldTime := now.Add(-48 * time.Hour).Format(time.RFC3339Nano) + newTime := now.Add(-30 * time.Minute).Format(time.RFC3339Nano) + + store.packets = []*StoreTx{ + { + ID: 1, + Hash: "old-hash", + FirstSeen: oldTime, + Observations: []*StoreObs{ + {ID: 1, PathJSON: `["abc"]`}, + }, + }, + { + ID: 2, + Hash: "new-hash", + FirstSeen: newTime, + Observations: []*StoreObs{ + {ID: 2, PathJSON: `["def"]`}, + }, + }, + } + + // With a 1-hour window, only the new tx should be processed. + // backfillResolvedPathsAsync will find no prefix map and finish quickly, + // but we can verify the pending count reflects the window. + go backfillResolvedPathsAsync(store, "", 100, time.Millisecond, 1) + + // Wait for completion + for i := 0; i < 100; i++ { + if store.backfillComplete.Load() { + break + } + time.Sleep(10 * time.Millisecond) + } + + if !store.backfillComplete.Load() { + t.Fatal("backfill did not complete") + } + + // With no prefix map, total should be 0 (early exit) or just the new one + // The function exits early when pm == nil, so backfillTotal stays at 0 + // if there were pending items but no pm. Let's verify it didn't process + // the old one by checking total <= 1. + total := store.backfillTotal.Load() + if total > 1 { + t.Errorf("backfill total = %d, want <= 1 (old tx should be excluded by hour window)", total) + } +} diff --git a/cmd/server/main.go b/cmd/server/main.go index 914a9185..48177487 100644 --- a/cmd/server/main.go +++ b/cmd/server/main.go @@ -153,7 +153,7 @@ func main() { // NOTE on startup ordering (review item #10): ensureResolvedPathColumn runs AFTER // OpenDB/detectSchema, so db.hasResolvedPath will be false on first run with a // pre-existing DB. This means Load() won't SELECT resolved_path from SQLite. - // That's OK: backfillResolvedPaths (below) computes and persists them in-memory + // Async backfill runs after HTTP starts (see backfillResolvedPathsAsync below) // AND to SQLite. On next restart, detectSchema finds the column and Load() reads it. if err := ensureResolvedPathColumn(dbPath); err != nil { log.Printf("[store] warning: could not add resolved_path column: %v", err) @@ -166,27 +166,59 @@ func main() { store.graph = loadNeighborEdgesFromDB(database.conn) log.Printf("[neighbor] loaded persisted neighbor graph") } else { - log.Printf("[neighbor] no persisted edges found, building from store...") - rw, rwErr := openRW(dbPath) - if rwErr == nil { - edgeCount := buildAndPersistEdges(store, rw) - rw.Close() - log.Printf("[neighbor] persisted %d edges", edgeCount) + log.Printf("[neighbor] no persisted edges found, will build in background...") + store.graph = NewNeighborGraph() // empty graph — gets populated by background goroutine + go func() { + defer func() { + if r := recover(); r != nil { + log.Printf("[neighbor] graph build panic recovered: %v", r) + } + }() + rw, rwErr := openRW(dbPath) + if rwErr == nil { + edgeCount := buildAndPersistEdges(store, rw) + rw.Close() + log.Printf("[neighbor] persisted %d edges", edgeCount) + } + built := BuildFromStore(store) + store.mu.Lock() + store.graph = built + store.mu.Unlock() + log.Printf("[neighbor] graph build complete") + }() + } + + // Initial pickBestObservation runs in background — doesn't need to block HTTP. + // API serves best-effort data until this completes (~10s for 100K txs). + // Processes in chunks of 5000, releasing the lock between chunks so API + // handlers remain responsive. + go func() { + defer func() { + if r := recover(); r != nil { + log.Printf("[store] pickBestObservation panic recovered: %v", r) + } + }() + const chunkSize = 5000 + store.mu.RLock() + totalPackets := len(store.packets) + store.mu.RUnlock() + + for i := 0; i < totalPackets; i += chunkSize { + end := i + chunkSize + if end > totalPackets { + end = totalPackets + } + store.mu.Lock() + for j := i; j < end && j < len(store.packets); j++ { + pickBestObservation(store.packets[j]) + } + store.mu.Unlock() + if end < totalPackets { + time.Sleep(10 * time.Millisecond) // yield to API handlers + } } - store.graph = BuildFromStore(store) - } - - // Backfill resolved_path for observations that don't have it yet - if backfilled := backfillResolvedPaths(store, dbPath); backfilled > 0 { - log.Printf("[store] backfilled resolved_path for %d observations", backfilled) - } - - // Re-pick best observation now that resolved paths are populated - store.mu.Lock() - for _, tx := range store.packets { - pickBestObservation(tx) - } - store.mu.Unlock() + log.Printf("[store] initial pickBestObservation complete (%d transmissions)", totalPackets) + }() // WebSocket hub hub := NewHub() @@ -234,6 +266,11 @@ func main() { close(pruneDone) } go func() { + defer func() { + if r := recover(); r != nil { + log.Printf("[prune] panic recovered: %v", r) + } + }() time.Sleep(1 * time.Minute) if n, err := database.PruneOldPackets(days); err != nil { log.Printf("[prune] error: %v", err) @@ -267,6 +304,11 @@ func main() { close(metricsPruneDone) } go func() { + defer func() { + if r := recover(); r != nil { + log.Printf("[metrics-prune] panic recovered: %v", r) + } + }() time.Sleep(2 * time.Minute) // stagger after packet prune database.PruneOldMetrics(metricsDays) for { @@ -281,6 +323,42 @@ func main() { log.Printf("[metrics-prune] auto-prune enabled: metrics older than %d days", metricsDays) } + // Auto-prune old neighbor edges + var stopEdgePrune func() + { + maxAgeDays := cfg.NeighborMaxAgeDays() + edgePruneTicker := time.NewTicker(24 * time.Hour) + edgePruneDone := make(chan struct{}) + stopEdgePrune = func() { + edgePruneTicker.Stop() + close(edgePruneDone) + } + go func() { + defer func() { + if r := recover(); r != nil { + log.Printf("[neighbor-prune] panic recovered: %v", r) + } + }() + time.Sleep(4 * time.Minute) // stagger after metrics prune + store.mu.RLock() + g := store.graph + store.mu.RUnlock() + PruneNeighborEdges(dbPath, g, maxAgeDays) + for { + select { + case <-edgePruneTicker.C: + store.mu.RLock() + g := store.graph + store.mu.RUnlock() + PruneNeighborEdges(dbPath, g, maxAgeDays) + case <-edgePruneDone: + return + } + } + }() + log.Printf("[neighbor-prune] auto-prune enabled: edges older than %d days", maxAgeDays) + } + // Graceful shutdown httpServer := &http.Server{ Addr: fmt.Sprintf(":%d", cfg.Port), @@ -306,6 +384,9 @@ func main() { if stopMetricsPrune != nil { stopMetricsPrune() } + if stopEdgePrune != nil { + stopEdgePrune() + } // 2. Gracefully drain HTTP connections (up to 15s) ctx, cancel := context.WithTimeout(context.Background(), 15*time.Second) @@ -325,6 +406,10 @@ func main() { }() log.Printf("[server] CoreScope (Go) listening on http://localhost:%d", cfg.Port) + + // Start async backfill in background — HTTP is now available. + go backfillResolvedPathsAsync(store, dbPath, 5000, 100*time.Millisecond, cfg.BackfillHours()) + if err := httpServer.ListenAndServe(); err != http.ErrServerClosed { log.Fatalf("[server] %v", err) } diff --git a/cmd/server/neighbor_graph.go b/cmd/server/neighbor_graph.go index 1bc93bad..25b6f378 100644 --- a/cmd/server/neighbor_graph.go +++ b/cmd/server/neighbor_graph.go @@ -542,3 +542,24 @@ func minLen(s string, n int) int { } return n } + +// PruneOlderThan removes all edges with LastSeen before cutoff. +// Returns the number of edges removed. +func (g *NeighborGraph) PruneOlderThan(cutoff time.Time) int { + g.mu.Lock() + defer g.mu.Unlock() + + pruned := 0 + for key, edge := range g.edges { + if edge.LastSeen.Before(cutoff) { + // Remove from byNode index + g.removeFromByNode(edge.NodeA, edge) + if edge.NodeB != "" { + g.removeFromByNode(edge.NodeB, edge) + } + delete(g.edges, key) + pruned++ + } + } + return pruned +} diff --git a/cmd/server/neighbor_persist.go b/cmd/server/neighbor_persist.go index 94f8dc1e..dfa0e664 100644 --- a/cmd/server/neighbor_persist.go +++ b/cmd/server/neighbor_persist.go @@ -343,112 +343,175 @@ func unmarshalResolvedPath(s string) []*string { return result } -// backfillResolvedPaths resolves paths for all observations that have NULL resolved_path. -func backfillResolvedPaths(store *PacketStore, dbPath string) int { - // Collect pending observations and snapshot immutable fields under read lock. - // graph is set in main.go before backfill is called; nil-safe throughout (review item #6). + +// backfillResolvedPathsAsync processes observations with NULL resolved_path in +// chunks, yielding between batches so HTTP handlers remain responsive. It sets +// store.backfillComplete when finished and re-picks best observations for any +// transmissions affected by newly resolved paths. +func backfillResolvedPathsAsync(store *PacketStore, dbPath string, chunkSize int, yieldDuration time.Duration, backfillHours int) { + defer func() { + if r := recover(); r != nil { + log.Printf("[store] backfillResolvedPathsAsync panic recovered: %v", r) + } + }() + // Collect ALL pending obs refs upfront in one pass under a single RLock (fix A). type obsRef struct { - obsID int - pathJSON string - observerID string - txJSON string // snapshot of DecodedJSON for extractFromNode + obsID int + pathJSON string + observerID string + txJSON string payloadType *int + txHash string // to re-pick best obs } + + cutoff := time.Now().UTC().Add(-time.Duration(backfillHours) * time.Hour) + store.mu.RLock() pm := store.nodePM - graph := store.graph - var pending []obsRef + var allPending []obsRef for _, tx := range store.packets { + // Skip transmissions older than the backfill window. + if tx.FirstSeen != "" { + if ts, err := time.Parse(time.RFC3339Nano, tx.FirstSeen); err == nil && ts.Before(cutoff) { + continue + } + // Also try the common SQLite format + if ts, err := time.Parse("2006-01-02 15:04:05", tx.FirstSeen); err == nil && ts.Before(cutoff) { + continue + } + } for _, obs := range tx.Observations { if obs.ResolvedPath == nil && obs.PathJSON != "" && obs.PathJSON != "[]" { - pending = append(pending, obsRef{ + allPending = append(allPending, obsRef{ obsID: obs.ID, pathJSON: obs.PathJSON, observerID: obs.ObserverID, txJSON: tx.DecodedJSON, payloadType: tx.PayloadType, + txHash: tx.Hash, }) } } } store.mu.RUnlock() - if len(pending) == 0 || pm == nil { - return 0 + totalPending := len(allPending) + if totalPending == 0 || pm == nil { + store.backfillComplete.Store(true) + log.Printf("[store] async resolved_path backfill: nothing to do") + return } - // Resolve paths outside the lock — resolvePathForObs only reads pm and graph. - type resolved struct { - obsID int - rp []*string - rpJSON string + store.backfillTotal.Store(int64(totalPending)) + store.backfillProcessed.Store(0) + log.Printf("[store] async resolved_path backfill starting: %d observations", totalPending) + + // Open RW connection once before the chunk loop (fix B). + var rw *sql.DB + if dbPath != "" { + var err error + rw, err = openRW(dbPath) + if err != nil { + log.Printf("[store] async backfill: open rw error: %v", err) + } } - var results []resolved - for _, ref := range pending { - // Build a minimal StoreTx for extractFromNode (only needs DecodedJSON + PayloadType). - fakeTx := &StoreTx{DecodedJSON: ref.txJSON, PayloadType: ref.payloadType} - rp := resolvePathForObs(ref.pathJSON, ref.observerID, fakeTx, pm, graph) - if len(rp) > 0 { - rpJSON := marshalResolvedPath(rp) - if rpJSON != "" { - results = append(results, resolved{ref.obsID, rp, rpJSON}) + defer func() { + if rw != nil { + rw.Close() + } + }() + + totalProcessed := 0 + for totalProcessed < totalPending { + end := totalProcessed + chunkSize + if end > totalPending { + end = totalPending + } + chunk := allPending[totalProcessed:end] + + // Re-read graph under RLock at the start of each chunk so we pick up + // a freshly-built graph once the background build goroutine completes, + // instead of using the potentially-empty graph captured at cold start. + store.mu.RLock() + graph := store.graph + store.mu.RUnlock() + + // Resolve paths outside any lock. + type resolved struct { + obsID int + rp []*string + rpJSON string + txHash string + } + var results []resolved + for _, ref := range chunk { + fakeTx := &StoreTx{DecodedJSON: ref.txJSON, PayloadType: ref.payloadType} + rp := resolvePathForObs(ref.pathJSON, ref.observerID, fakeTx, pm, graph) + if len(rp) > 0 { + rpJSON := marshalResolvedPath(rp) + if rpJSON != "" { + results = append(results, resolved{ref.obsID, rp, rpJSON, ref.txHash}) + } } } - } - if len(results) == 0 { - return 0 - } + // Persist to SQLite using the shared connection. + if len(results) > 0 && rw != nil { + sqlTx, err := rw.Begin() + if err != nil { + log.Printf("[store] async backfill: begin tx error: %v", err) + } else { + stmt, err := sqlTx.Prepare("UPDATE observations SET resolved_path = ? WHERE id = ?") + if err != nil { + log.Printf("[store] async backfill: prepare error: %v", err) + sqlTx.Rollback() + } else { + var execErr error + for _, r := range results { + if _, e := stmt.Exec(r.rpJSON, r.obsID); e != nil && execErr == nil { + execErr = e + } + } + if execErr != nil { + log.Printf("[store] async backfill: exec error (first): %v", execErr) + } + stmt.Close() + if err := sqlTx.Commit(); err != nil { + log.Printf("[store] async backfill: commit error: %v", err) + } + } + } - // Persist to SQLite (no lock needed — separate RW connection). - rw, err := openRW(dbPath) - if err != nil { - log.Printf("[store] backfill: open rw error: %v", err) - return 0 - } - defer rw.Close() - - sqlTx, err := rw.Begin() - if err != nil { - log.Printf("[store] backfill: begin tx error: %v", err) - return 0 - } - defer sqlTx.Rollback() - - stmt, err := sqlTx.Prepare("UPDATE observations SET resolved_path = ? WHERE id = ?") - if err != nil { - log.Printf("[store] backfill: prepare error: %v", err) - return 0 - } - defer stmt.Close() - - var firstErr error - for _, r := range results { - if _, err := stmt.Exec(r.rpJSON, r.obsID); err != nil && firstErr == nil { - firstErr = err + // Update in-memory state and re-pick best observation under a single + // write lock. The per-tx pickBestObservation is O(observations) which is + // typically <10 per tx — negligible cost vs. the race risk of splitting + // the lock (pollAndMerge can append to tx.Observations concurrently). + store.mu.Lock() + affectedSet := make(map[string]bool) + for _, r := range results { + if obs, ok := store.byObsID[r.obsID]; ok { + obs.ResolvedPath = r.rp + } + if !affectedSet[r.txHash] { + affectedSet[r.txHash] = true + if tx, ok := store.byHash[r.txHash]; ok { + pickBestObservation(tx) + } + } + } + store.mu.Unlock() } - } - if firstErr != nil { - log.Printf("[store] backfill resolved_path exec error (first): %v", firstErr) + + totalProcessed += len(chunk) + store.backfillProcessed.Store(int64(totalProcessed)) + pct := float64(totalProcessed) / float64(totalPending) * 100 + log.Printf("[store] backfill progress: %d/%d observations (%.1f%%)", totalProcessed, totalPending, pct) + + time.Sleep(yieldDuration) } - if err := sqlTx.Commit(); err != nil { - log.Printf("[store] backfill: commit error: %v", err) - return 0 - } - - // Update in-memory state under write lock. - store.mu.Lock() - count := 0 - for _, r := range results { - if obs, ok := store.byObsID[r.obsID]; ok { - obs.ResolvedPath = r.rp - count++ - } - } - store.mu.Unlock() - - return count + store.backfillComplete.Store(true) + log.Printf("[store] async resolved_path backfill complete: %d observations processed", totalProcessed) } // ─── Shared helpers ──────────────────────────────────────────────────────────── @@ -529,3 +592,34 @@ func openRW(dbPath string) (*sql.DB, error) { rw.SetMaxOpenConns(1) return rw, nil } + +// PruneNeighborEdges removes edges older than maxAgeDays from both SQLite and +// the in-memory graph. Uses openRW internally because the shared database.conn +// is opened with mode=ro and DELETEs would silently fail. +func PruneNeighborEdges(dbPath string, graph *NeighborGraph, maxAgeDays int) (int, error) { + cutoff := time.Now().UTC().Add(-time.Duration(maxAgeDays) * 24 * time.Hour) + + // 1. Prune from SQLite using a read-write connection + var dbPruned int64 + rw, err := openRW(dbPath) + if err != nil { + return 0, fmt.Errorf("prune neighbor_edges: open rw: %w", err) + } + defer rw.Close() + res, err := rw.Exec("DELETE FROM neighbor_edges WHERE last_seen < ?", cutoff.Format(time.RFC3339)) + if err != nil { + return 0, fmt.Errorf("prune neighbor_edges: %w", err) + } + dbPruned, _ = res.RowsAffected() + + // 2. Prune from in-memory graph + memPruned := 0 + if graph != nil { + memPruned = graph.PruneOlderThan(cutoff) + } + + if dbPruned > 0 || memPruned > 0 { + log.Printf("[neighbor-prune] removed %d DB rows, %d in-memory edges older than %d days", dbPruned, memPruned, maxAgeDays) + } + return int(dbPruned), nil +} diff --git a/cmd/server/routes.go b/cmd/server/routes.go index 859dcefb..f0030962 100644 --- a/cmd/server/routes.go +++ b/cmd/server/routes.go @@ -101,6 +101,9 @@ func (s *Server) RegisterRoutes(r *mux.Router) { // Performance instrumentation middleware r.Use(s.perfMiddleware) + // Backfill status header middleware + r.Use(s.backfillStatusMiddleware) + // Config endpoints r.HandleFunc("/api/config/cache", s.handleConfigCache).Methods("GET") r.HandleFunc("/api/config/client", s.handleConfigClient).Methods("GET") @@ -164,6 +167,17 @@ func (s *Server) RegisterRoutes(r *mux.Router) { r.HandleFunc("/api/audio-lab/buckets", s.handleAudioLabBuckets).Methods("GET") } +func (s *Server) backfillStatusMiddleware(next http.Handler) http.Handler { + return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + if s.store != nil && s.store.backfillComplete.Load() { + w.Header().Set("X-CoreScope-Status", "ready") + } else { + w.Header().Set("X-CoreScope-Status", "backfilling") + } + next.ServeHTTP(w, r) + }) +} + func (s *Server) perfMiddleware(next http.Handler) http.Handler { return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { if !strings.HasPrefix(r.URL.Path, "/api/") { @@ -521,6 +535,19 @@ func (s *Server) handleStats(w http.ResponseWriter, r *http.Request) { return } counts := s.db.GetRoleCounts() + + // Compute backfill progress + backfilling := s.store != nil && !s.store.backfillComplete.Load() + var backfillProgress float64 + if backfilling && s.store != nil && s.store.backfillTotal.Load() > 0 { + backfillProgress = float64(s.store.backfillProcessed.Load()) / float64(s.store.backfillTotal.Load()) + if backfillProgress > 1 { + backfillProgress = 1 + } + } else if !backfilling { + backfillProgress = 1 + } + resp := &StatsResponse{ TotalPackets: stats.TotalPackets, TotalTransmissions: &stats.TotalTransmissions, @@ -540,6 +567,8 @@ func (s *Server) handleStats(w http.ResponseWriter, r *http.Request) { Companions: counts["companions"], Sensors: counts["sensors"], }, + Backfilling: backfilling, + BackfillProgress: backfillProgress, } s.statsMu.Lock() diff --git a/cmd/server/store.go b/cmd/server/store.go index 334d8242..2bf1eb39 100644 --- a/cmd/server/store.go +++ b/cmd/server/store.go @@ -157,6 +157,12 @@ type PacketStore struct { // Persisted neighbor graph for hop resolution at ingest time. graph *NeighborGraph + // Async backfill state: set after backfillResolvedPathsAsync completes. + backfillComplete atomic.Bool + // Progress tracking for async backfill (total pending and processed so far). + backfillTotal atomic.Int64 // set once at start of async backfill + backfillProcessed atomic.Int64 + // Eviction config and stats retentionHours float64 // 0 = unlimited maxMemoryMB int // 0 = unlimited diff --git a/cmd/server/types.go b/cmd/server/types.go index 6b6fbceb..5f350d53 100644 --- a/cmd/server/types.go +++ b/cmd/server/types.go @@ -68,6 +68,8 @@ type StatsResponse struct { Commit string `json:"commit"` BuildTime string `json:"buildTime"` Counts RoleCounts `json:"counts"` + Backfilling bool `json:"backfilling"` + BackfillProgress float64 `json:"backfillProgress"` } // ─── Health ──────────────────────────────────────────────────────────────────── diff --git a/config.example.json b/config.example.json index e8864ef9..b31e0d4d 100644 --- a/config.example.json +++ b/config.example.json @@ -8,13 +8,15 @@ }, "https": { "cert": "/path/to/cert.pem", - "key": "/path/to/key.pem" + "key": "/path/to/key.pem", + "_comment": "TLS cert/key paths for direct HTTPS. Most deployments use Caddy (included in Docker) for auto-TLS instead." }, "branding": { "siteName": "CoreScope", "tagline": "Real-time MeshCore LoRa mesh network analyzer", "logoUrl": null, - "faviconUrl": null + "faviconUrl": null, + "_comment": "Customize site name, tagline, logo, and favicon. logoUrl/faviconUrl can be absolute URLs or relative paths." }, "theme": { "accent": "#4a9eff", @@ -23,38 +25,75 @@ "navBg2": "#1a1a2e", "statusGreen": "#45644c", "statusYellow": "#b08b2d", - "statusRed": "#b54a4a" + "statusRed": "#b54a4a", + "_comment": "CSS color overrides. Use the in-app Theme Customizer for live preview, then export values here." }, "nodeColors": { "repeater": "#dc2626", "companion": "#2563eb", "room": "#16a34a", "sensor": "#d97706", - "observer": "#8b5cf6" + "observer": "#8b5cf6", + "_comment": "Marker/badge colors per node role. Used on map, nodes list, and live feed." }, "home": { "heroTitle": "CoreScope", "heroSubtitle": "Find your nodes to start monitoring them.", "steps": [ - { "emoji": "📡", "title": "Connect", "description": "Link your node to the mesh" }, - { "emoji": "🔍", "title": "Monitor", "description": "Watch packets flow in real-time" }, - { "emoji": "📊", "title": "Analyze", "description": "Understand your network's health" } + { + "emoji": "\ud83d\udce1", + "title": "Connect", + "description": "Link your node to the mesh" + }, + { + "emoji": "\ud83d\udd0d", + "title": "Monitor", + "description": "Watch packets flow in real-time" + }, + { + "emoji": "\ud83d\udcca", + "title": "Analyze", + "description": "Understand your network's health" + } ], "checklist": [ - { "question": "How do I add my node?", "answer": "Search for your node name or paste your public key." }, - { "question": "What regions are covered?", "answer": "Check the map page to see active observers and nodes." } + { + "question": "How do I add my node?", + "answer": "Search for your node name or paste your public key." + }, + { + "question": "What regions are covered?", + "answer": "Check the map page to see active observers and nodes." + } ], "footerLinks": [ - { "label": "📦 Packets", "url": "#/packets" }, - { "label": "🗺️ Network Map", "url": "#/map" }, - { "label": "🔴 Live", "url": "#/live" }, - { "label": "📡 All Nodes", "url": "#/nodes" }, - { "label": "💬 Channels", "url": "#/channels" } - ] + { + "label": "\ud83d\udce6 Packets", + "url": "#/packets" + }, + { + "label": "\ud83d\uddfa\ufe0f Network Map", + "url": "#/map" + }, + { + "label": "\ud83d\udd34 Live", + "url": "#/live" + }, + { + "label": "\ud83d\udce1 All Nodes", + "url": "#/nodes" + }, + { + "label": "\ud83d\udcac Channels", + "url": "#/channels" + } + ], + "_comment": "Customize the landing page hero, onboarding steps, FAQ, and footer links." }, "mqtt": { "broker": "mqtt://localhost:1883", - "topic": "meshcore/+/+/packets" + "topic": "meshcore/+/+/packets", + "_comment": "Legacy single-broker config. Prefer mqttSources[] for multiple brokers." }, "mqttSources": [ { @@ -150,11 +189,26 @@ "timezone": "local", "formatPreset": "iso", "customFormat": "", - "allowCustomFormat": false + "allowCustomFormat": false, + "_comment": "defaultMode: ago|local|iso. timezone: local|utc. formatPreset: iso|us|eu. customFormat: strftime-style (requires allowCustomFormat: true)." }, "packetStore": { "maxMemoryMB": 1024, "estimatedPacketBytes": 450, "_comment": "In-memory packet store. maxMemoryMB caps RAM usage. All packets loaded on startup, served from RAM." - } -} \ No newline at end of file + }, + "resolvedPath": { + "backfillHours": 24, + "_comment": "How far back (hours) the async backfill scans for observations with NULL resolved_path. Default: 24. Set higher to backfill older data, lower to speed up startup." + }, + "neighborGraph": { + "maxAgeDays": 5, + "_comment": "Neighbor edges older than this many days are pruned on startup and daily. Default: 5." + }, + "_comment_mqttSources": "Each source connects to an MQTT broker. topics: what to subscribe to. iataFilter: only ingest packets from these regions (optional).", + "_comment_channelKeys": "Hex keys for decrypting channel messages. Key name = channel display name. public channel key is well-known.", + "_comment_hashChannels": "Channel names whose keys are derived via SHA256. Key = SHA256(name)[:16]. Listed here so the ingestor can auto-derive keys.", + "_comment_defaultRegion": "IATA code shown by default in region filters.", + "_comment_mapDefaults": "Initial map center [lat, lon] and zoom level.", + "_comment_regions": "IATA code to display name mapping. Packets are tagged with region codes by MQTT topic structure." +}