perf: use prepared statements for frequently-called server DB queries (rebase of #1878) (#1934)

Continues #1878 by @Joel-Claw, at their request. Both commits are theirs, authorship unchanged; I only rebased them onto master and resolved the conflict with #1909. ## The conflict, and how it is resolved Exactly the two places I named in the review on #1878: `OpenDB` and `Close()`. Both PRs rewrite them, and #1909 went first because it is the correctness fix. **`OpenDB`** — kept #1909's pinned-connection `detectSchema` and added this PR's `prepareStatements()` after it: ```go derr := d.detectSchema(ctx, sc) _ = sc.Close() if derr != nil { conn.Close(); return nil, fmt.Errorf("schema detection failed: %w", derr) } // Statements are prepared after schema detection so they can never be // compiled against a schema mode that turned out to be wrong (#1901). if err := d.prepareStatements(); err != nil { ... } ``` The ordering matters and is not arbitrary: preparing before detection would compile statements against a schema mode that #1909 exists to stop trusting. **`Close()`** — kept this PR's statement closing and **did not** restore the WAL checkpoint. #1909 removed it deliberately: the handle is `mode=ro`, so `PRAGMA wal_checkpoint(TRUNCATE)` can only ever fail with "disk I/O error (778)" and was emitting a misleading storage-fault line on every shutdown. That reasoning survives; the statement closing is added in front of it. ## Verification - Both commits cherry-picked onto `e5595ad9` - `cmd/server` builds - **Full `cmd/server` suite: ok, 0 failures** (not just the targeted DB tests — after master briefly went red today from a two-PR interaction, a full local run seemed worth the two minutes) ## Review points still open, none blocking From my review on #1878, unchanged by the rebase: 1. Every SQL string now exists twice, once prepared and once as the `stmtQueryRow` fallback literal, with nothing keeping them in sync. The fallback is genuinely needed — twelve test helpers build `&DB{conn: ...}` directly and never call `prepareStatements` — but a constructor for those helpers would remove the duplication. 2. `stmtCountObsLastHour` and `stmtCountObsLastDay` are byte-identical SQL. 3. `OpenDB` now refuses to start rather than degrading when a Prepare fails. Contained today, since none of the 13 prepared queries touch a schema-conditional column, but the failure mode changed. @Joel-Claw — your work, your credit. Ping me if you would rather take it back. 🤖 Generated with [Claude Code](https://claude.com/claude-code) https://claude.ai/code/session_01Wzwr3eXseyNM7Xj598djjE --------- Co-authored-by: Joel Claw <358739783+Joel-Claw@users.noreply.github.com>
2026-09-14 07:26:07 +00:00 · 2026-09-02 14:34:16 +02:00
parent aabbd50d27
commit 2f711eb851
1 changed files with 123 additions and 54 deletions
@@ -21,7 +21,9 @@ import (
 // routeTypeTransport covers TRANSPORT_FLOOD (0) and TRANSPORT_DIRECT (3) —
 // the only route types that carry transport_code_1 (transport-level scope).
 // Per firmware/docs/packet_format.md § Route Types:
-//   0 = TRANSPORT_FLOOD, 1 = FLOOD, 2 = DIRECT, 3 = TRANSPORT_DIRECT.
+//
+//	0 = TRANSPORT_FLOOD, 1 = FLOOD, 2 = DIRECT, 3 = TRANSPORT_DIRECT.
+//
 // Routes 1 (FLOOD) and 2 (DIRECT) never carry a scope by protocol — they are
 // inherently unscoped and are counted separately in GetScopeStats (#1838).
 const routeTypeTransportSQL = "route_type IN (0, 3)"
@@ -33,11 +35,11 @@ const routeTypeNonTransportSQL = "route_type IN (1, 2)"

 // DB wraps a read-only connection to the MeshCore SQLite database.
 type DB struct {
-	conn             *sql.DB
-	path             string // filesystem path to the database file
-	isV3             bool   // v3 schema: observer_idx in observations (vs observer_id in v2)
-	hasResolvedPath  bool   // observations table has resolved_path column
-	hasObsRawHex     bool   // observations table has raw_hex column (#881)
+	conn                *sql.DB
+	path                string // filesystem path to the database file
+	isV3                bool   // v3 schema: observer_idx in observations (vs observer_id in v2)
+	hasResolvedPath     bool   // observations table has resolved_path column
+	hasObsRawHex        bool   // observations table has raw_hex column (#881)
 	hasScopeName        bool   // transmissions.scope_name column exists (#899)
 	hasDefaultScope     bool   // nodes.default_scope column exists (#899)
 	hasMultibyteSupCols bool   // nodes/inactive_nodes have multibyte_sup/multibyte_evidence (#903)
@@ -48,6 +50,22 @@ type DB struct {
 	channelsCacheKey string
 	channelsCacheRes []map[string]interface{}
 	channelsCacheExp time.Time
+
+	// Prepared statements for frequently-called queries.
+	// Prepared once at initDB time, reused across all requests.
+	stmtCountTransmissions  *sql.Stmt
+	stmtCountObservations   *sql.Stmt
+	stmtCountNodesActive    *sql.Stmt // WHERE last_seen > ?
+	stmtCountNodesAll       *sql.Stmt
+	stmtCountObservers      *sql.Stmt
+	stmtCountObsLastHour    *sql.Stmt // WHERE timestamp > ?
+	stmtCountObsLastDay     *sql.Stmt // WHERE timestamp > ?
+	stmtNodeLookup          *sql.Stmt // WHERE public_key = ? OR name = ?
+	stmtTxByHash            *sql.Stmt // WHERE hash = ?
+	stmtCountNodesByRole    *sql.Stmt // WHERE role = ? AND last_seen > ?
+	stmtCountNodesByRoleAll *sql.Stmt // WHERE role = ?
+	stmtMaxTxID             *sql.Stmt // COALESCE(MAX(id), 0) FROM transmissions
+	stmtMaxObsID            *sql.Stmt // COALESCE(MAX(id), 0) FROM observations
 }

 // OpenDB opens a read-only SQLite connection with WAL mode.
@@ -82,15 +100,74 @@ func OpenDB(path string) (*DB, error) {
 		conn.Close()
 		return nil, fmt.Errorf("schema detection failed: %w", derr)
 	}
+	// Statements are prepared after schema detection so they can never be
+	// compiled against a schema mode that turned out to be wrong (#1901).
+	if err := d.prepareStatements(); err != nil {
+		conn.Close()
+		return nil, fmt.Errorf("prepare statements: %w", err)
+	}
 	return d, nil
 }

+// stmtQueryRow returns a QueryRow-like helper that uses the prepared statement
+// when non-nil (production), or falls back to a direct db.conn.QueryRow query
+// when the statement is nil (test DBs that skip prepareStatements).
+func (db *DB) stmtQueryRow(stmt *sql.Stmt, fallbackSQL string, args ...interface{}) *sql.Row {
+	if stmt != nil {
+		return stmt.QueryRow(args...)
+	}
+	return db.conn.QueryRow(fallbackSQL, args...)
+}
+
+// prepareStatements creates prepared statements for frequently-called queries.
+// SQLite compiles and caches the query plan once, avoiding re-parsing per request.
+func (db *DB) prepareStatements() error {
+	var err error
+	prepare := func(query string) *sql.Stmt {
+		if err != nil {
+			return nil
+		}
+		var s *sql.Stmt
+		s, err = db.conn.Prepare(query)
+		return s
+	}
+
+	db.stmtCountTransmissions = prepare("SELECT COUNT(*) FROM transmissions")
+	db.stmtCountObservations = prepare("SELECT COUNT(*) FROM observations")
+	db.stmtCountNodesActive = prepare("SELECT COUNT(*) FROM nodes WHERE last_seen > ?")
+	db.stmtCountNodesAll = prepare("SELECT COUNT(*) FROM nodes")
+	db.stmtCountObservers = prepare("SELECT COUNT(*) FROM observers WHERE inactive IS NULL OR inactive = 0")
+	db.stmtCountObsLastHour = prepare("SELECT COUNT(*) FROM observations WHERE timestamp > ?")
+	db.stmtCountObsLastDay = prepare("SELECT COUNT(*) FROM observations WHERE timestamp > ?")
+	db.stmtNodeLookup = prepare("SELECT public_key FROM nodes WHERE public_key = ? OR name = ? LIMIT 1")
+	db.stmtTxByHash = prepare("SELECT id FROM transmissions WHERE hash = ?")
+	db.stmtCountNodesByRole = prepare("SELECT COUNT(*) FROM nodes WHERE role = ? AND last_seen > ?")
+	db.stmtCountNodesByRoleAll = prepare("SELECT COUNT(*) FROM nodes WHERE role = ?")
+	db.stmtMaxTxID = prepare("SELECT COALESCE(MAX(id), 0) FROM transmissions")
+	db.stmtMaxObsID = prepare("SELECT COALESCE(MAX(id), 0) FROM observations")
+	return err
+}
+
 func (db *DB) Close() error {
-	// The connection is opened read-only (mode=ro in OpenDB), so it cannot
-	// checkpoint the WAL: PRAGMA wal_checkpoint(TRUNCATE) always fails with
-	// "disk I/O error (778)" on a read-only handle. Attempting it emitted a
-	// misleading storage-fault line on every shutdown that wasted incident
-	// investigation time (#1901); the ingestor (the writer) owns checkpointing.
+	// Close prepared statements
+	closers := []*sql.Stmt{
+		db.stmtCountTransmissions, db.stmtCountObservations,
+		db.stmtCountNodesActive, db.stmtCountNodesAll, db.stmtCountObservers,
+		db.stmtCountObsLastHour, db.stmtCountObsLastDay,
+		db.stmtNodeLookup, db.stmtTxByHash,
+		db.stmtCountNodesByRole, db.stmtCountNodesByRoleAll,
+		db.stmtMaxTxID, db.stmtMaxObsID,
+	}
+	for _, s := range closers {
+		if s != nil {
+			s.Close()
+		}
+	}
+	// No WAL checkpoint here. The connection is opened read-only (mode=ro in
+	// OpenDB), so PRAGMA wal_checkpoint(TRUNCATE) always fails with "disk I/O
+	// error (778)" on it. Attempting it emitted a misleading storage-fault line
+	// on every shutdown that wasted incident investigation time (#1901); the
+	// ingestor (the writer) owns checkpointing.
 	return db.conn.Close()
 }

@@ -248,7 +325,7 @@ func (db *DB) scanTransmissionRow(rows *sql.Rows) map[string]interface{} {

 // Node represents a row from the nodes table.
 type Node struct {
-	PublicKey     string   `json:"public_key"`
+	PublicKey    string   `json:"public_key"`
 	Name         *string  `json:"name"`
 	Role         *string  `json:"role"`
 	Lat          *float64 `json:"lat"`
@@ -341,24 +418,24 @@ type Stats struct {
 // GetStats returns aggregate counts (matches Node.js db.getStats shape).
 func (db *DB) GetStats() (*Stats, error) {
 	s := &Stats{}
-	err := db.conn.QueryRow("SELECT COUNT(*) FROM transmissions").Scan(&s.TotalTransmissions)
+	err := db.stmtQueryRow(db.stmtCountTransmissions, "SELECT COUNT(*) FROM transmissions").Scan(&s.TotalTransmissions)
 	if err != nil {
 		return nil, err
 	}
 	s.TotalPackets = s.TotalTransmissions

-	db.conn.QueryRow("SELECT COUNT(*) FROM observations").Scan(&s.TotalObservations)
+	db.stmtQueryRow(db.stmtCountObservations, "SELECT COUNT(*) FROM observations").Scan(&s.TotalObservations)
 	// Node.js uses 7-day active nodes for totalNodes
 	sevenDaysAgo := time.Now().Add(-7 * 24 * time.Hour).Format(time.RFC3339)
-	db.conn.QueryRow("SELECT COUNT(*) FROM nodes WHERE last_seen > ?", sevenDaysAgo).Scan(&s.TotalNodes)
-	db.conn.QueryRow("SELECT COUNT(*) FROM nodes").Scan(&s.TotalNodesAllTime)
-	db.conn.QueryRow("SELECT COUNT(*) FROM observers WHERE inactive IS NULL OR inactive = 0").Scan(&s.TotalObservers)
+	db.stmtQueryRow(db.stmtCountNodesActive, "SELECT COUNT(*) FROM nodes WHERE last_seen > ?", sevenDaysAgo).Scan(&s.TotalNodes)
+	db.stmtQueryRow(db.stmtCountNodesAll, "SELECT COUNT(*) FROM nodes").Scan(&s.TotalNodesAllTime)
+	db.stmtQueryRow(db.stmtCountObservers, "SELECT COUNT(*) FROM observers WHERE inactive IS NULL OR inactive = 0").Scan(&s.TotalObservers)

 	oneHourAgo := time.Now().Add(-1 * time.Hour).Unix()
-	db.conn.QueryRow("SELECT COUNT(*) FROM observations WHERE timestamp > ?", oneHourAgo).Scan(&s.PacketsLastHour)
+	db.stmtQueryRow(db.stmtCountObsLastHour, "SELECT COUNT(*) FROM observations WHERE timestamp > ?", oneHourAgo).Scan(&s.PacketsLastHour)

 	oneDayAgo := time.Now().Add(-24 * time.Hour).Unix()
-	db.conn.QueryRow("SELECT COUNT(*) FROM observations WHERE timestamp > ?", oneDayAgo).Scan(&s.PacketsLast24h)
+	db.stmtQueryRow(db.stmtCountObsLastDay, "SELECT COUNT(*) FROM observations WHERE timestamp > ?", oneDayAgo).Scan(&s.PacketsLast24h)

 	return s, nil
 }
@@ -480,7 +557,7 @@ func (db *DB) GetRoleCounts() map[string]int {
 	counts := map[string]int{}
 	for _, role := range []string{"repeater", "room", "companion", "sensor"} {
 		var c int
-		db.conn.QueryRow("SELECT COUNT(*) FROM nodes WHERE role = ? AND last_seen > ?", role, sevenDaysAgo).Scan(&c)
+		db.stmtQueryRow(db.stmtCountNodesByRole, "SELECT COUNT(*) FROM nodes WHERE role = ? AND last_seen > ?", role, sevenDaysAgo).Scan(&c)
 		counts[role+"s"] = c
 	}
 	return counts
@@ -491,7 +568,7 @@ func (db *DB) GetAllRoleCounts() map[string]int {
 	counts := map[string]int{}
 	for _, role := range []string{"repeater", "room", "companion", "sensor"} {
 		var c int
-		db.conn.QueryRow("SELECT COUNT(*) FROM nodes WHERE role = ?", role).Scan(&c)
+		db.stmtQueryRow(db.stmtCountNodesByRoleAll, "SELECT COUNT(*) FROM nodes WHERE role = ?", role).Scan(&c)
 		counts[role+"s"] = c
 	}
 	return counts
@@ -499,20 +576,20 @@ func (db *DB) GetAllRoleCounts() map[string]int {

 // PacketQuery holds filter params for packet listing.
 type PacketQuery struct {
-	Limit    int
-	Offset   int
-	Type     *int
-	Route    *int
-	Observer string
-	Hash     string
-	Since    string
-	Until    string
-	Region   string
-	Area     string   // area key; filters by transmitting node's GPS position
-	Node     string
-	Channel  string // channel_hash filter (#812). Plain names like "#test"/"public" or "enc_<HEX>" for encrypted
-	Order               string // ASC or DESC
-	ExpandObservations  bool   // when true, include observation sub-maps in txToMap output
+	Limit              int
+	Offset             int
+	Type               *int
+	Route              *int
+	Observer           string
+	Hash               string
+	Since              string
+	Until              string
+	Region             string
+	Area               string // area key; filters by transmitting node's GPS position
+	Node               string
+	Channel            string // channel_hash filter (#812). Plain names like "#test"/"public" or "enc_<HEX>" for encrypted
+	Order              string // ASC or DESC
+	ExpandObservations bool   // when true, include observation sub-maps in txToMap output
 }

 // PacketResult wraps paginated packet list.
@@ -541,7 +618,7 @@ func (db *DB) QueryPackets(q PacketQuery) (*PacketResult, error) {
 	// Count transmissions (not observations)
 	var total int
 	if len(where) == 0 {
-		db.conn.QueryRow("SELECT COUNT(*) FROM transmissions").Scan(&total)
+		db.stmtQueryRow(db.stmtCountTransmissions, "SELECT COUNT(*) FROM transmissions").Scan(&total)
 	} else {
 		countSQL := fmt.Sprintf("SELECT COUNT(*) FROM transmissions t %s", w)
 		db.conn.QueryRow(countSQL, args...).Scan(&total)
@@ -593,7 +670,7 @@ func (db *DB) QueryGroupedPackets(q PacketQuery) (*PacketResult, error) {
 	// Count total transmissions (fast — queries transmissions directly, not a VIEW)
 	var total int
 	if len(where) == 0 {
-		db.conn.QueryRow("SELECT COUNT(*) FROM transmissions").Scan(&total)
+		db.stmtQueryRow(db.stmtCountTransmissions, "SELECT COUNT(*) FROM transmissions").Scan(&total)
 	} else {
 		db.conn.QueryRow(fmt.Sprintf("SELECT COUNT(*) FROM transmissions t %s", w), args...).Scan(&total)
 	}
@@ -825,14 +902,13 @@ func (db *DB) buildTransmissionWhere(q PacketQuery) ([]string, []interface{}) {

 func (db *DB) resolveNodePubkey(nodeIDOrName string) string {
 	var pk string
-	err := db.conn.QueryRow("SELECT public_key FROM nodes WHERE public_key = ? OR name = ? LIMIT 1", nodeIDOrName, nodeIDOrName).Scan(&pk)
+	err := db.stmtQueryRow(db.stmtNodeLookup, "SELECT public_key FROM nodes WHERE public_key = ? OR name = ? LIMIT 1", nodeIDOrName, nodeIDOrName).Scan(&pk)
 	if err != nil {
 		return nodeIDOrName
 	}
 	return pk
 }

-
 // GetTransmissionByID fetches from transmissions table with observer data.
 func (db *DB) GetTransmissionByID(id int) (map[string]interface{}, error) {
 	selectCols, observerJoin := db.transmissionBaseSQL()
@@ -870,8 +946,7 @@ func (db *DB) GetPacketByHash(hash string) (map[string]interface{}, error) {
 // when the in-memory PacketStore has pruned the entry but the DB still has it.
 func (db *DB) GetObservationsForHash(hash string) []map[string]interface{} {
 	var txID int
-	err := db.conn.QueryRow("SELECT id FROM transmissions WHERE hash = ?",
-		strings.ToLower(hash)).Scan(&txID)
+	err := db.stmtQueryRow(db.stmtTxByHash, "SELECT id FROM transmissions WHERE hash = ?", strings.ToLower(hash)).Scan(&txID)
 	if err != nil {
 		return nil
 	}
@@ -879,7 +954,6 @@ func (db *DB) GetObservationsForHash(hash string) []map[string]interface{} {
 	return obsByTx[txID]
 }

-
 // GetNodes returns filtered, paginated node list.
 func (db *DB) GetNodes(limit, offset int, role, search, before, lastHeard, sortBy, region string) ([]map[string]interface{}, int, map[string]int, error) {
 	var where []string
@@ -1059,7 +1133,6 @@ func (db *DB) GetNodeByPubkey(pubkey string) (map[string]interface{}, error) {
 	return nil, nil
 }

-
 // GetRecentTransmissionsForNode returns recent transmissions originated by a
 // node, identified by exact pubkey match on the indexed from_pubkey column
 // (#1143). The legacy `name` substring fallback was removed: it produced
@@ -1432,7 +1505,6 @@ func (db *DB) GetDistinctIATAs() ([]string, error) {
 	return codes, nil
 }

-
 // GetNetworkStatus returns overall network health status.
 func (db *DB) GetNetworkStatus(healthThresholds HealthThresholds) (map[string]interface{}, error) {
 	rows, err := db.conn.Query("SELECT public_key, name, role, last_seen FROM nodes")
@@ -1900,8 +1972,8 @@ func (db *DB) GetChannelMessages(channelHash string, limit, offset int, region .
 	defer rows.Close()

 	type msg struct {
-		Data       map[string]interface{}
-		Repeats    int
+		Data        map[string]interface{}
+		Repeats     int
 		LatestEpoch int64 // max observation timestamp (unix seconds) — issue #1366
 	}
 	msgMap := make(map[int]*msg, len(pageIDs))
@@ -2016,8 +2088,6 @@ func (db *DB) GetChannelMessages(channelHash string, limit, offset int, region .
 	return messages, total, nil
 }

-
-
 // GetNewTransmissionsSince returns new transmissions after a given ID for WebSocket polling.
 func (db *DB) GetNewTransmissionsSince(lastID int, limit int) ([]map[string]interface{}, error) {
 	if limit <= 0 {
@@ -2053,14 +2123,14 @@ func (db *DB) GetNewTransmissionsSince(lastID int, limit int) ([]map[string]inte
 // GetMaxTransmissionID returns the current max ID for polling.
 func (db *DB) GetMaxTransmissionID() int {
 	var maxID int
-	db.conn.QueryRow("SELECT COALESCE(MAX(id), 0) FROM transmissions").Scan(&maxID)
+	db.stmtQueryRow(db.stmtMaxTxID, "SELECT COALESCE(MAX(id), 0) FROM transmissions").Scan(&maxID)
 	return maxID
 }

 // GetMaxObservationID returns the current max observation ID for polling.
 func (db *DB) GetMaxObservationID() int {
 	var maxID int
-	db.conn.QueryRow("SELECT COALESCE(MAX(id), 0) FROM observations").Scan(&maxID)
+	db.stmtQueryRow(db.stmtMaxObsID, "SELECT COALESCE(MAX(id), 0) FROM observations").Scan(&maxID)
 	return maxID
 }

@@ -2657,7 +2727,6 @@ func (db *DB) GetMetricsSummary(since string) ([]MetricsSummaryRow, error) {
 // (PruneOldMetrics / RemoveStaleObservers removed in #1283 — see note
 // above the MetricsSample type. Ingestor owns these writes now.)

-
 // GetDroppedPackets returns recently dropped packets, newest first.
 func (db *DB) GetDroppedPackets(limit int, observerID, nodePubkey string) ([]map[string]interface{}, error) {
 	if limit <= 0 || limit > 500 {
@@ -2797,7 +2866,7 @@ func (db *DB) GetScopeStats(window string) (*ScopeStatsResponse, error) {
 			COALESCE(SUM(CASE WHEN scope_name IS NULL THEN 1 ELSE 0 END), 0) AS unscoped,
 			COALESCE(SUM(CASE WHEN scope_name = '' THEN 1 ELSE 0 END), 0) AS unknown_scope
 		FROM transmissions
-		WHERE ` + routeTypeTransportSQL + ` AND first_seen >= ?
+		WHERE `+routeTypeTransportSQL+` AND first_seen >= ?
 	`, since)
 	if err := row.Scan(
 		&resp.Summary.TransportTotal,
@@ -2825,7 +2894,7 @@ func (db *DB) GetScopeStats(window string) (*ScopeStatsResponse, error) {
 	rows, err := db.conn.Query(`
 		SELECT scope_name, COUNT(*) AS cnt
 		FROM transmissions
-		WHERE ` + routeTypeTransportSQL + ` AND scope_name IS NOT NULL AND scope_name != '' AND first_seen >= ?
+		WHERE `+routeTypeTransportSQL+` AND scope_name IS NOT NULL AND scope_name != '' AND first_seen >= ?
 		GROUP BY scope_name
 		ORDER BY cnt DESC
 	`, since)
@@ -2852,7 +2921,7 @@ func (db *DB) GetScopeStats(window string) (*ScopeStatsResponse, error) {
 			COUNT(scope_name) AS scoped,
 			SUM(CASE WHEN scope_name IS NULL THEN 1 ELSE 0 END) AS unscoped
 		FROM transmissions
-		WHERE ` + routeTypeTransportSQL + ` AND first_seen >= ?
+		WHERE `+routeTypeTransportSQL+` AND first_seen >= ?
 		GROUP BY bucket
 		ORDER BY bucket
 	`, bucketExpr)