fix(dbschema): stop gating the dedup repair on an error string, and log what it deletes

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

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

cmd/server/prune-requests/request-085fc40c889b4543.json

@@ -1,8 +0,0 @@
-{
-  "id": "085fc40c889b4543",
-  "requestedAt": "2026-09-09T20:39:40.398005Z",
-  "reason": "geo-prune",
-  "pubkeys": [
-    "aaaa111122223333"
-  ]
-}

cmd/server/prune-requests/request-b74000055aa4c213.json

@@ -1,8 +0,0 @@
-{
-  "id": "b74000055aa4c213",
-  "requestedAt": "2026-09-09T20:46:44.481208Z",
-  "reason": "geo-prune",
-  "pubkeys": [
-    "aaaa111122223333"
-  ]
-}

cmd/server/prune-requests/result-cc924dfc7c5711ae.json

@@ -1,6 +0,0 @@
-{
-  "id": "cc924dfc7c5711ae",
-  "requestedAt": "2026-09-09T20:46:43.491617Z",
-  "completedAt": "2026-09-09T20:46:44.491617Z",
-  "deleted": 1
-}

cmd/server/prune-requests/result-df9a2e4ef5563fef.json

@@ -1,6 +0,0 @@
-{
-  "id": "df9a2e4ef5563fef",
-  "requestedAt": "2026-09-09T20:39:39.409166Z",
-  "completedAt": "2026-09-09T20:39:40.409167Z",
-  "deleted": 1
-}

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

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

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