mirror of
https://github.com/Kpa-clawbot/meshcore-analyzer.git
synced 2026-10-23 11:32:23 +00:00
refactor(db): finish #1283 — Option 4: ingestor owns neighbor-graph + schema migrations; server is read-only (fixes #1287) (#1289)
Red commit: https://github.com/Kpa-clawbot/CoreScope/commit/eae179b99b5fd34924547632aa8f8025c405aa53 (CI: pending — opens with this PR) Finishes #1283. RED test `TestServerSourceHasNoCachedRWCalls` goes from failing (13 writer call-sites) to GREEN (zero). Per #1287 Option 4 (https://github.com/Kpa-clawbot/CoreScope/issues/1287#issuecomment-4485099992): ingestor owns the neighbor graph build + persist; server reads the snapshot. **Category A — Schema migrations** → new `internal/dbschema` package. `dbschema.Apply(rw)` runs in `cmd/ingestor` startup (in `OpenStore`). `dbschema.AssertReady(ro)` runs in `cmd/server/main.go` and FATAL-LOG-EXITS if any expected column/index/table is missing — the operator must restart the ingestor first. Covers indexes, `neighbor_edges`, `observations.resolved_path`, `observers.{inactive,last_packet_at,iata}`, `(inactive_)nodes.foreign_advert`, `transmissions.from_pubkey`. **Category B — Backfill** → ingestor. `BackfillFromPubkey` and observer-blacklist soft-delete moved to `cmd/ingestor/maintenance.go`. Server keeps an inert `fromPubkeyBackfillSnapshot` stub for `/api/healthz` API compatibility. **Category C — Neighbor-graph persistence (Option 4)** → ingestor writes, server reads. - Ingestor (`cmd/ingestor/neighbor_builder.go`): every 60s scans `observations + transmissions`, extracts edges (originator↔first-hop for ADVERTs; observer↔last-hop for all), resolves hop prefixes via a node-table prefix index, upserts into `neighbor_edges`. - Server (`cmd/server/neighbor_recomputer.go`): every 60s re-reads `neighbor_edges` and atomic-swaps the resulting `NeighborGraph` into `s.graph`. Initial load is synchronous on startup. All server-side incremental edge writers (the two `asyncPersistResolvedPathsAndEdges` paths in `cmd/server/store.go`) are gone. - Neighbor-edge daily prune (`PruneNeighborEdges`) moved to ingestor. **Why Option 4**: clean read/write separation, no startup CPU spike (server loads existing snapshot instead of rebuilding from history), no IPC/delta-protocol churn. Staleness budget ~60s — same model as the analytics recomputers in #1240 / #1248 / #672 axis 2. **Recomputer interval default for neighbor graph**: 60s (`NeighborGraphRecomputerDefaultInterval`, `NeighborEdgesBuilderInterval`). **Invariants added**: - `TestServerSourceHasNoCachedRWCalls` (RED commit eae179b9): grep enforces zero `cachedRW(`, `mode=rw`, or `sql.Open(_journal_mode=WAL…)` in non-test `cmd/server/` sources. - `TestServerStartupRequiresMigratedSchema`: server refuses to start against an unmigrated DB. - `TestNeighborGraphRecomputerLoadsSnapshot`: post-write snapshot is picked up on the next refresh. - `TestNeighborEdgesBuilderUpsertsFromObservations`: end-to-end pipeline writes the expected edge. `grep cachedRW cmd/server/*.go | grep -v _test.go` → 0 matches. Fixes #1287. --------- Co-authored-by: MeshCore Bot <bot@meshcore.local> Co-authored-by: Kpa-clawbot <Kpa-clawbot@users.noreply.github.com> Co-authored-by: corescope-bot <bot@corescope.local>
This commit is contained in:
31 files changed
+1514
-2862
No files matched your search
@@ -1,12 +1,8 @@
|
||||
package main
|
||||
|
||||
import (
|
||||
"database/sql"
|
||||
"path/filepath"
|
||||
"testing"
|
||||
"time"
|
||||
|
||||
_ "modernc.org/sqlite"
|
||||
)
|
||||
|
||||
func TestBackfillHoursDefault(t *testing.T) {
|
||||
@@ -72,106 +68,3 @@ func TestGraphPruneOlderThan(t *testing.T) {
|
||||
}
|
||||
}
|
||||
|
||||
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)
|
||||
}
|
||||
}
|
||||
Reference in new issue
Block a user