From ac7e8d38a2d9a390e4681c9ff794b67ff1781869 Mon Sep 17 00:00:00 2001 From: Sylvain Rabot Date: Fri, 11 Sep 2026 11:29:42 +0200 Subject: [PATCH] fix(dbschema): stop gating the dedup repair on an error string, and log what it deletes MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Review feedback on #1992, all three points on `internal/dbschema/dedup_index.go`. **The repair decision no longer hinges on prose.** `ensureObservationsDedupIndex` matched `strings.Contains(err.Error(), "UNIQUE constraint failed")` to decide whether to repair — a text match on driver output, introduced by the same change that swapped the driver. A reworded or wrapped message would skip the repair silently and resurface as an `OpenStore` failure with nothing pointing at the cause. It now checks `sqlite3.Error.Code == sqlite3.ErrConstraint` via `errors.As`. Confirmed against the driver: mattn returns `Code=19` (`ErrConstraint`), `ExtendedCode=2067` (`ErrConstraintUnique`) for `CREATE UNIQUE INDEX` over duplicates, and `errors.As` reaches it. `TestEnsureObservationsDedupIndexTakesRepairPathOnRealDriverError` covers the branch that *decides* to repair, which nothing did: the existing tests call `collapseDuplicatesAndIndex` directly, so a broken error check would have left all of them green. It asserts `isConstraintViolation` recognises the driver's own error and that the repair actually runs. **The TEMP table lifecycle now matches its comment.** The deferred cleanup ran `rw.Exec("DROP TABLE IF EXISTS temp.dedup_groups")`, but a TEMP table belongs to one connection and `database/sql` hands out pooled ones, so that drop could land on a different connection and leave the table on the one holding it. Harmless in `cmd/ingestor` at `SetMaxOpenConns(1)`, not guaranteed for `cmd/migrate`, which calls the same function with an unbounded pool. Every reference is now through `tx`: created, dropped before COMMIT, and removed by ROLLBACK on the error paths. Verified rather than assumed — a TEMP table created inside a transaction does not survive its rollback, so the deferred drop was both unnecessary and aimed at the wrong connection. The leading `DROP ... IF EXISTS` stays, since this package cannot prove nothing else left one behind. **The deletion is auditable now.** `collapsed N duplicate observation row(s)` was the whole record, and if the merge direction were ever wrong again that line is all anyone would have to work from. The group keys are sitting in `dedup_groups` at that moment, so they get logged — group count, rows to remove, then the first 20 groups with their key and the id being kept — before anything is destroyed. It also warns up front that the write lock is held, because on a large table the pause at ingestor startup is otherwise unexplained: [dbschema] idx_observations_dedup cannot be created: duplicate observations exist. Repairing now — this holds the write lock until it completes, and on a large observations table it can take tens of seconds. [dbschema] 1 duplicate observation group(s), 1 row(s) to remove [dbschema] transmission_id=4 observer_idx=4 path_json="[]" rows=2 keeping id=1 [dbschema] collapsed 1 duplicate observation row(s); idx_observations_dedup created **Docs carry the production numbers.** The PR's headline came from a standalone harness; @efiten ran both drivers against an 11M-observation, 9.7GB instance on 4-core arm64, with the order reversed in a second round so the page cache favoured the old driver. Warm, the 7d audit is ~1.8x and chunk load ~1.4x, and `/api/nodes?limit=500` is unchanged. So the real gain on the paths that matter is 1.4-1.8x, not the 2-2.3x the harness showed, and the doc now says to quote those instead. It also records the counterweight nobody had quantified: a cold native build goes from 52s to 163s, which an instance building its own image pays per deploy. **Also removes four files the previous commit should not have added.** `cmd/server/prune-requests/*.json` is runtime queue state written by `internal/prunequeue` and left behind by the server tests; a `git add -A` swept it in. They were never in master, so the branch's net diff was unaffected, but they had no business being committed. `prune-requests/` is now in `.gitignore` so it cannot happen again. Constraint: the repair must stay all-or-nothing — merge, delete and CREATE UNIQUE INDEX share one transaction Rejected: keeping the string match with a test pinning the message | the typed code is available and the text is the driver's to change Rejected: an explicit sql.Conn for the TEMP table | tx already pins one connection, and rollback already cleans up Rejected: logging every duplicate group | unbounded output at startup is its own operational problem, so it caps at 20 with a count of the rest Directive: nothing in this file may decide control flow from an error message; use the typed code Directive: every TEMP table reference goes through tx, never rw — the pool will hand you a different connection Confidence: high Scope-risk: narrow Not-tested: the collapse at 11M rows, or against a database an ingestor is actively writing to — @efiten has offered a staging instance taking live MQTT traffic, which is the remaining gap before this leaves draft --- .gitignore | 3 + .../request-085fc40c889b4543.json | 8 -- .../request-b74000055aa4c213.json | 8 -- .../result-cc924dfc7c5711ae.json | 6 - .../result-df9a2e4ef5563fef.json | 6 - docs/sqlite-driver-migration.md | 37 ++++++ internal/dbschema/dedup_index.go | 109 ++++++++++++++++-- internal/dbschema/dedup_index_test.go | 79 ++++++++++++- 8 files changed, 216 insertions(+), 40 deletions(-) delete mode 100644 cmd/server/prune-requests/request-085fc40c889b4543.json delete mode 100644 cmd/server/prune-requests/request-b74000055aa4c213.json delete mode 100644 cmd/server/prune-requests/result-cc924dfc7c5711ae.json delete mode 100644 cmd/server/prune-requests/result-df9a2e4ef5563fef.json diff --git a/.gitignore b/.gitignore index 74a964cc..33def577 100644 --- a/.gitignore +++ b/.gitignore @@ -11,6 +11,9 @@ theme.json firmware/ coverage/ dist/ +# internal/prunequeue writes these next to the database; the server tests leave +# them in cmd/server/. Runtime queue state, never repo content. +prune-requests/ public-instrumented/ .nyc_output/ .setup-state diff --git a/cmd/server/prune-requests/request-085fc40c889b4543.json b/cmd/server/prune-requests/request-085fc40c889b4543.json deleted file mode 100644 index c11afdbf..00000000 --- a/cmd/server/prune-requests/request-085fc40c889b4543.json +++ /dev/null @@ -1,8 +0,0 @@ -{ - "id": "085fc40c889b4543", - "requestedAt": "2026-09-09T20:39:40.398005Z", - "reason": "geo-prune", - "pubkeys": [ - "aaaa111122223333" - ] -} \ No newline at end of file diff --git a/cmd/server/prune-requests/request-b74000055aa4c213.json b/cmd/server/prune-requests/request-b74000055aa4c213.json deleted file mode 100644 index e1b192fc..00000000 --- a/cmd/server/prune-requests/request-b74000055aa4c213.json +++ /dev/null @@ -1,8 +0,0 @@ -{ - "id": "b74000055aa4c213", - "requestedAt": "2026-09-09T20:46:44.481208Z", - "reason": "geo-prune", - "pubkeys": [ - "aaaa111122223333" - ] -} \ No newline at end of file diff --git a/cmd/server/prune-requests/result-cc924dfc7c5711ae.json b/cmd/server/prune-requests/result-cc924dfc7c5711ae.json deleted file mode 100644 index bfc58bd1..00000000 --- a/cmd/server/prune-requests/result-cc924dfc7c5711ae.json +++ /dev/null @@ -1,6 +0,0 @@ -{ - "id": "cc924dfc7c5711ae", - "requestedAt": "2026-09-09T20:46:43.491617Z", - "completedAt": "2026-09-09T20:46:44.491617Z", - "deleted": 1 -} \ No newline at end of file diff --git a/cmd/server/prune-requests/result-df9a2e4ef5563fef.json b/cmd/server/prune-requests/result-df9a2e4ef5563fef.json deleted file mode 100644 index 0fdfdccc..00000000 --- a/cmd/server/prune-requests/result-df9a2e4ef5563fef.json +++ /dev/null @@ -1,6 +0,0 @@ -{ - "id": "df9a2e4ef5563fef", - "requestedAt": "2026-09-09T20:39:39.409166Z", - "completedAt": "2026-09-09T20:39:40.409167Z", - "deleted": 1 -} \ No newline at end of file diff --git a/docs/sqlite-driver-migration.md b/docs/sqlite-driver-migration.md index 0164016e..9bf5f8ad 100644 --- a/docs/sqlite-driver-migration.md +++ b/docs/sqlite-driver-migration.md @@ -27,6 +27,32 @@ corescope's own hot-path SQL under both drivers (Apple M4, `-count=5`, medians): Allocation counts drop with it: 1.12M vs 1.64M allocs (−32%) and 21 MB vs 30 MB (−30%) on the chunk load, 20.3k vs 27.3k on the lookups. +## Confirmed on production + +The table above comes from a standalone harness. @efiten then ran both drivers +against a real instance — 11,077,038 observations, 9.7 GB database, 4-core arm64 +— as server-only containers against the same live volume, one at a time. Round 2 +reverses the order so the page-cache advantage goes to the old driver: + +| | audit 7d | audit 24h | background fill (13 chunks) | start → /api/health | +|---|---:|---:|---:|---:| +| modernc, round 1 | 16.67s | 2.27s | 130.2s | 16.6s | +| mattn, round 1 | 7.87s | 1.34s | 93.8s | 13.5s | +| mattn, round 2 | 8.15s | 1.35s | 96.4s | 13.0s | +| modernc, round 2 | 13.46s | 2.29s | 137.8s | 15.5s | + +Warm, the old driver improves to 13.46s on the 7d audit and still loses by +~1.8×. Chunk load is ~1.4×. `/api/nodes?limit=500` is 0.039s against 0.037s — +i.e. nothing. + +**Real-world gains are smaller than the harness suggests: ~1.4-1.8× on the +paths that matter, not 2-2.3×.** The shape holds — scans and joins gain, small +lookups do not — but quote these numbers, not the harness ones. + +**Build time is the counterweight**, cold and native on that machine: +**52s on master, 163s on this branch**. An instance that builds its own image +pays that on every deploy. + ## The cost: the build is cgo now `CGO_ENABLED=0` still *builds*, which is the trap: mattn links a stub, and the @@ -102,6 +128,17 @@ Merge, delete and index creation all share one transaction. Split apart, a writer inserting a duplicate in the gap makes the index creation fail while the deletions stay committed — rows destroyed and no index to show for it. +The repair logs what it is about to destroy — group count, rows to remove, and +the first 20 group keys with the id it keeps — before deleting anything, because +a bare row count is not enough to reconstruct from if the merge direction is +ever wrong again. It also says up front that it holds the write lock, since on a +large table that pause at ingestor startup is otherwise unexplained. + +The cost is measured at 4.1s on 2.4M synthetic rows holding 5 duplicates. That +is well short of a real deployment: an 11M-row instance has not been measured, +and the collapse has not been run against a database an ingestor is actively +writing to. + Note `COALESCE(path_json, '')` makes a NULL path and an empty-string path the same key, but `NULL` and `'[]'` different keys. See `internal/dbschema/dedup_index_test.go`, where the last-wins and atomicity diff --git a/internal/dbschema/dedup_index.go b/internal/dbschema/dedup_index.go index f751aba7..727ee7de 100644 --- a/internal/dbschema/dedup_index.go +++ b/internal/dbschema/dedup_index.go @@ -2,8 +2,10 @@ package dbschema import ( "database/sql" + "errors" "fmt" - "strings" + + sqlite3 "github.com/mattn/go-sqlite3" ) // upsertMergedColumns are the columns the ingestor's observation UPSERT writes @@ -39,7 +41,8 @@ var upsertMergedColumns = []string{"snr", "rssi", "score", "raw_hex", "resolved_ // transaction and holds the write lock for its duration, so the repetition is // worth removing. const dupGroupsDDL = `CREATE TEMP TABLE dedup_groups AS - SELECT transmission_id, observer_idx, COALESCE(path_json, '') AS p, MIN(id) AS keep + SELECT transmission_id, observer_idx, COALESCE(path_json, '') AS p, + MIN(id) AS keep, COUNT(*) AS n FROM observations GROUP BY transmission_id, observer_idx, COALESCE(path_json, '') HAVING COUNT(*) > 1` @@ -82,18 +85,43 @@ func ensureObservationsDedupIndex(rw *sql.DB, logf Logger) error { if err == nil { return nil } - if !strings.Contains(err.Error(), "UNIQUE constraint failed") { + if !isConstraintViolation(err) { return err } - removed, err := collapseDuplicatesAndIndex(rw) + // Deleting rows is not something to do quietly, and on a large table this + // holds the write lock long enough that an operator watching startup + // deserves to know why before it happens rather than after. + logf("[dbschema] idx_observations_dedup cannot be created: duplicate observations exist. " + + "Repairing now — this holds the write lock until it completes, and on a large " + + "observations table it can take tens of seconds.") + + removed, err := collapseDuplicatesAndIndex(rw, logf) if err != nil { return fmt.Errorf("collapse duplicate observations: %w", err) } - logf("[dbschema] collapsed %d duplicate observation row(s) so idx_observations_dedup could be created", removed) + logf("[dbschema] collapsed %d duplicate observation row(s); idx_observations_dedup created", removed) return nil } +// isConstraintViolation reports whether err is SQLite refusing a constraint. +// +// Deliberately a typed check. This used to match on +// strings.Contains(err, "UNIQUE constraint failed"), which made the entire +// repair path — including the row deletion — hinge on the driver's prose, in +// the same change that swapped the driver. A reworded or wrapped message would +// silently skip the repair and surface later as an OpenStore failure with no +// hint as to why. CREATE UNIQUE INDEX can only hit a constraint error because +// the data violates the uniqueness it asks for, so the primary code is the +// right granularity. +func isConstraintViolation(err error) bool { + var se sqlite3.Error + if errors.As(err, &se) { + return se.Code == sqlite3.ErrConstraint + } + return false +} + // collapseDuplicatesAndIndex merges duplicate observation rows into the lowest // id of each group, deletes the rest, and creates the unique index — all in one // transaction. It returns the number of rows deleted. @@ -102,29 +130,41 @@ func ensureObservationsDedupIndex(rw *sql.DB, logf Logger) error { // them separated, a writer inserting a duplicate in the gap makes the index // creation fail while leaving the deletions committed: rows destroyed and no // index to show for it. One transaction makes the repair all-or-nothing. -func collapseDuplicatesAndIndex(rw *sql.DB) (int64, error) { +func collapseDuplicatesAndIndex(rw *sql.DB, logf Logger) (int64, error) { tx, err := rw.Begin() if err != nil { return 0, err } defer func() { _ = tx.Rollback() }() - // A TEMP table lives for the life of the connection, and database/sql hands - // out pooled connections, so an earlier call may have left one behind on - // this one. Drop before creating rather than assume, and again afterwards - // so the connection returns to the pool clean. + // Every reference to the TEMP table goes through tx, so its whole lifecycle + // sits on one connection: created here, dropped before COMMIT below, and + // removed by ROLLBACK on any error path (a TEMP table created inside a + // transaction does not survive its rollback — verified, not assumed). + // + // It has to be tx, not rw. A TEMP table belongs to a single connection and + // database/sql hands out pooled ones, so an `rw.Exec` drop can land on a + // different connection and leave the table behind on the one that has it. + // The leading DROP is still here because this package cannot prove nothing + // else ever left one on this connection. if _, err := tx.Exec(`DROP TABLE IF EXISTS temp.dedup_groups`); err != nil { return 0, fmt.Errorf("drop stale dedup_groups: %w", err) } if _, err := tx.Exec(dupGroupsDDL); err != nil { return 0, fmt.Errorf("build dedup_groups: %w", err) } - defer func() { _, _ = rw.Exec(`DROP TABLE IF EXISTS temp.dedup_groups`) }() if _, err := tx.Exec(`CREATE INDEX temp.dedup_groups_key ON dedup_groups(transmission_id, observer_idx, p)`); err != nil { return 0, fmt.Errorf("index dedup_groups: %w", err) } + // The group keys exist right now and cease to after the delete. Record them + // before destroying anything: a row count is not enough to reconstruct from + // if the merge direction is ever wrong again. + if err := logDuplicateGroups(tx, logf); err != nil { + return 0, err + } + // Only merge columns this database actually has. Apply runs this step // before ensureResolvedPathColumn and ensureObservationsRawHexColumn, so on // a database old enough to be missing the dedup index, resolved_path and @@ -168,6 +208,10 @@ func collapseDuplicatesAndIndex(rw *sql.DB) (int64, error) { } // Same transaction: if this fails, the deletions above roll back with it. + if _, err := tx.Exec(`DROP TABLE temp.dedup_groups`); err != nil { + return 0, fmt.Errorf("drop dedup_groups: %w", err) + } + if _, err := tx.Exec(dedupIndexDDL); err != nil { return 0, fmt.Errorf("create idx_observations_dedup after collapsing %d row(s): %w", removed, err) } @@ -192,3 +236,46 @@ func existingColumns(rw *sql.DB, table string, want []string) ([]string, error) } return out, nil } + +// maxLoggedDupGroups bounds the audit log. A database missing the index for +// long enough can have a great many duplicate groups, and an unbounded dump at +// startup is its own operational problem. +const maxLoggedDupGroups = 20 + +// logDuplicateGroups records what is about to be collapsed, while dedup_groups +// still holds it. Reads through tx because the TEMP table lives on that +// transaction's connection. +func logDuplicateGroups(tx *sql.Tx, logf Logger) error { + var groups, extra int64 + if err := tx.QueryRow(`SELECT COUNT(*), COALESCE(SUM(n - 1), 0) FROM dedup_groups`). + Scan(&groups, &extra); err != nil { + return fmt.Errorf("count dedup_groups: %w", err) + } + logf("[dbschema] %d duplicate observation group(s), %d row(s) to remove", groups, extra) + + rows, err := tx.Query(`SELECT transmission_id, observer_idx, p, keep, n + FROM dedup_groups ORDER BY keep LIMIT ?`, maxLoggedDupGroups) + if err != nil { + return fmt.Errorf("list dedup_groups: %w", err) + } + defer rows.Close() + var listed int64 + for rows.Next() { + var txID, keep, n int64 + var observerIdx sql.NullInt64 + var pathJSON string + if err := rows.Scan(&txID, &observerIdx, &pathJSON, &keep, &n); err != nil { + return fmt.Errorf("scan dedup_groups: %w", err) + } + logf("[dbschema] transmission_id=%d observer_idx=%v path_json=%q rows=%d keeping id=%d", + txID, observerIdx.Int64, pathJSON, n, keep) + listed++ + } + if err := rows.Err(); err != nil { + return fmt.Errorf("iterate dedup_groups: %w", err) + } + if groups > listed { + logf("[dbschema] ... and %d more group(s) not listed", groups-listed) + } + return nil +} diff --git a/internal/dbschema/dedup_index_test.go b/internal/dbschema/dedup_index_test.go index d0258a36..b572d603 100644 --- a/internal/dbschema/dedup_index_test.go +++ b/internal/dbschema/dedup_index_test.go @@ -2,7 +2,9 @@ package dbschema import ( "database/sql" + "fmt" "path/filepath" + "strings" "testing" _ "github.com/mattn/go-sqlite3" @@ -173,7 +175,7 @@ func TestCollapseDuplicatesAndIndexIsAtomic(t *testing.T) { t.Fatal(err) } - if _, err := collapseDuplicatesAndIndex(db); err == nil { + if _, err := collapseDuplicatesAndIndex(db, t.Logf); err == nil { t.Fatal("expected index creation to fail") } @@ -225,3 +227,78 @@ func TestEnsureObservationsDedupIndexSkipsV2Schema(t *testing.T) { t.Error("index must not be created on a v2 schema") } } + +// Nothing covered the branch that DECIDES to repair. TestCollapseDuplicates... +// calls collapseDuplicatesAndIndex directly, so a broken error check in +// ensureObservationsDedupIndex would leave every one of those tests green while +// production silently skipped the repair and failed later at OpenStore. +// +// This asserts the decision: duplicates present, the real driver's real error, +// and the repair actually taken. It is the test that would catch the driver +// rewording its constraint message. +func TestEnsureObservationsDedupIndexTakesRepairPathOnRealDriverError(t *testing.T) { + db := observationsDB(t) + if _, err := db.Exec(`INSERT INTO observations (id, transmission_id, observer_idx, path_json, timestamp) VALUES + (1, 4, 4, '[]', 10), (2, 4, 4, '[]', 10)`); err != nil { + t.Fatal(err) + } + + // The error the fast path actually gets. If isConstraintViolation stops + // recognising this, the repair below never runs. + _, createErr := db.Exec(dedupIndexDDL) + if createErr == nil { + t.Fatal("expected CREATE UNIQUE INDEX to fail over duplicates") + } + if !isConstraintViolation(createErr) { + t.Fatalf("isConstraintViolation did not recognise the driver's own error: %v (%T)", createErr, createErr) + } + + var repaired bool + logf := func(format string, args ...interface{}) { + repaired = true + t.Logf(format, args...) + } + if err := ensureObservationsDedupIndex(db, logf); err != nil { + t.Fatalf("ensureObservationsDedupIndex: %v", err) + } + if !repaired { + t.Error("repair path was not taken: no log output, so the constraint error was not recognised") + } + if !dedupIndexExists(t, db) { + t.Error("index missing after repair") + } + var n int + if err := db.QueryRow(`SELECT COUNT(*) FROM observations`).Scan(&n); err != nil { + t.Fatal(err) + } + if n != 1 { + t.Errorf("observations = %d, want 1", n) + } +} + +// The audit log has to name what it destroyed, not just count it. +func TestCollapseLogsGroupKeysBeforeDeleting(t *testing.T) { + db := observationsDB(t) + if _, err := db.Exec(`INSERT INTO observations (id, transmission_id, observer_idx, path_json, timestamp) VALUES + (1, 11, 3, '["AA"]', 10), (2, 11, 3, '["AA"]', 10), (3, 11, 3, '["AA"]', 10)`); err != nil { + t.Fatal(err) + } + var out []string + logf := func(format string, args ...interface{}) { + out = append(out, fmt.Sprintf(format, args...)) + } + if err := ensureObservationsDedupIndex(db, logf); err != nil { + t.Fatal(err) + } + joined := strings.Join(out, "\n") + for _, want := range []string{ + "1 duplicate observation group(s), 2 row(s) to remove", + "transmission_id=11", + `path_json="[\"AA\"]"`, + "keeping id=1", + } { + if !strings.Contains(joined, want) { + t.Errorf("audit log missing %q; got:\n%s", want, joined) + } + } +}