fix(indexstore): limit SQLite to single connection and add atomic replace with retry (#141)

GORM's default connection pool opened 22+ SQLite connections, causing
pragmas (WAL, busy_timeout) to only apply to one connection. Multiple
connections triggered SQLITE_IOERR_BEGIN_ATOMIC (6410) from VFS-level
write contention.

- Set SetMaxOpenConns(1) for SQLite in both indexstore and auth store
- Add ReplaceTestStats/ReplaceTestStatsBlockLogs for atomic
delete+insert in a single transaction (prevents data loss on insert
failure)
- Add withRetry helper for transient SQLite errors with exponential
backoff

Author: skylenet

pkg/api/indexer/indexer.go

@@ -496,11 +496,6 @@ func (idx *indexer) indexTestStats(
 	entry *executor.IndexEntry,
 	resultData []byte,
 ) error {
-	// Delete old test stats for this run before re-inserting.
-	if err := idx.store.DeleteTestStatsForRun(ctx, runID); err != nil {
-		return fmt.Errorf("deleting old test stats: %w", err)
-	}
-
 	// Use AccumulateRunResult to extract per-test durations.
 	suiteStats := make(executor.SuiteStats)
 
@@ -576,8 +571,8 @@ func (idx *indexer) indexTestStats(
 		}
 	}
 
-	if err := idx.store.BulkUpsertTestStats(ctx, testStats); err != nil {
-		return fmt.Errorf("bulk inserting test stats: %w", err)
+	if err := idx.store.ReplaceTestStats(ctx, runID, testStats); err != nil {
+		return fmt.Errorf("replacing test stats: %w", err)
 	}
 
 	return nil
@@ -645,11 +640,6 @@ func (idx *indexer) indexTestStatsBlockLogs(
 	suiteHash, runID, client string,
 	data []byte,
 ) error {
-	// Delete old block logs for this run before re-inserting.
-	if err := idx.store.DeleteTestStatsBlockLogsForRun(ctx, runID); err != nil {
-		return fmt.Errorf("deleting old test stats block logs: %w", err)
-	}
-
 	// The file is a map of test name -> single block log entry.
 	var testMap map[string]blockLogEntry
 	if err := json.Unmarshal(data, &testMap); err != nil {
@@ -698,8 +688,10 @@ func (idx *indexer) indexTestStatsBlockLogs(
 		})
 	}
 
-	if err := idx.store.BulkInsertTestStatsBlockLogs(ctx, logs); err != nil {
-		return fmt.Errorf("bulk inserting test stats block logs: %w", err)
+	if err := idx.store.ReplaceTestStatsBlockLogs(
+		ctx, runID, logs,
+	); err != nil {
+		return fmt.Errorf("replacing test stats block logs: %w", err)
 	}
 
 	return nil

pkg/api/indexstore/indexstore.go

@@ -3,6 +3,8 @@ package indexstore
 import (
 	"context"
 	"fmt"
+	"strings"
+	"time"
 
 	"github.com/ethpandaops/benchmarkoor/pkg/config"
 	"github.com/glebarez/sqlite"
@@ -28,6 +30,9 @@ type Store interface {
 	BulkUpsertTestStats(
 		ctx context.Context, stats []*TestStat,
 	) error
+	ReplaceTestStats(
+		ctx context.Context, runID string, stats []*TestStat,
+	) error
 	ListTestStatsBySuite(
 		ctx context.Context, suiteHash string,
 	) ([]TestStat, error)
@@ -45,6 +50,9 @@ type Store interface {
 	BulkInsertTestStatsBlockLogs(
 		ctx context.Context, logs []*TestStatsBlockLog,
 	) error
+	ReplaceTestStatsBlockLogs(
+		ctx context.Context, runID string, logs []*TestStatsBlockLog,
+	) error
 	DeleteTestStatsBlockLogsForRun(ctx context.Context, runID string) error
 
 	QueryRuns(ctx context.Context, params *QueryParams) (*QueryResult, error)
@@ -112,6 +120,18 @@ func (s *store) Start(ctx context.Context) error {
 
 	s.db = db
 
+	// SQLite requires a single connection to avoid write contention and
+	// ensure pragmas are applied consistently. GORM's default pool opens
+	// many connections, each with independent pragma state.
+	if s.cfg.Driver == "sqlite" {
+		sqlDB, err := db.DB()
+		if err != nil {
+			return fmt.Errorf("getting underlying sql.DB: %w", err)
+		}
+
+		sqlDB.SetMaxOpenConns(1)
+	}
+
 	// Set SQLite pragmas for performance and reliability.
 	if s.cfg.Driver == "sqlite" {
 		pragmas := []string{
@@ -418,6 +438,82 @@ func (s *store) DeleteTestStatsForRun(
 	return nil
 }
 
+// ReplaceTestStats atomically deletes old test stats for a run and inserts
+// new ones in a single transaction with retry for transient SQLite errors.
+func (s *store) ReplaceTestStats(
+	ctx context.Context, runID string, stats []*TestStat,
+) error {
+	return s.withRetry(func() error {
+		return s.db.WithContext(ctx).Transaction(func(tx *gorm.DB) error {
+			if err := tx.Where("run_id = ?", runID).
+				Delete(&TestStat{}).Error; err != nil {
+				return fmt.Errorf(
+					"deleting test stats for run: %w", err,
+				)
+			}
+
+			if len(stats) == 0 {
+				return nil
+			}
+
+			const batchSize = 100
+			for i := 0; i < len(stats); i += batchSize {
+				end := min(i+batchSize, len(stats))
+				batch := stats[i:end]
+
+				if err := tx.CreateInBatches(
+					batch, len(batch),
+				).Error; err != nil {
+					return fmt.Errorf(
+						"bulk inserting test stats: %w", err,
+					)
+				}
+			}
+
+			return nil
+		})
+	})
+}
+
+// ReplaceTestStatsBlockLogs atomically deletes old block logs for a run
+// and inserts new ones in a single transaction with retry.
+func (s *store) ReplaceTestStatsBlockLogs(
+	ctx context.Context, runID string, logs []*TestStatsBlockLog,
+) error {
+	return s.withRetry(func() error {
+		return s.db.WithContext(ctx).Transaction(func(tx *gorm.DB) error {
+			if err := tx.Where("run_id = ?", runID).
+				Delete(&TestStatsBlockLog{}).Error; err != nil {
+				return fmt.Errorf(
+					"deleting test stats block logs for run: %w",
+					err,
+				)
+			}
+
+			if len(logs) == 0 {
+				return nil
+			}
+
+			const batchSize = 100
+			for i := 0; i < len(logs); i += batchSize {
+				end := min(i+batchSize, len(logs))
+				batch := logs[i:end]
+
+				if err := tx.CreateInBatches(
+					batch, len(batch),
+				).Error; err != nil {
+					return fmt.Errorf(
+						"bulk inserting test stats block logs: %w",
+						err,
+					)
+				}
+			}
+
+			return nil
+		})
+	})
+}
+
 // BulkInsertTestStatsBlockLogs inserts multiple test stats block log records
 // in batches. Each batch is auto-committed independently to keep
 // WAL/journal pressure low. The caller deletes old records first.
@@ -638,6 +734,48 @@ func (s *store) QuerySuites(
 	}, nil
 }
 
+// withRetry retries fn up to 3 times on transient SQLite errors
+// (database locked, disk I/O) with exponential backoff.
+func (s *store) withRetry(fn func() error) error {
+	const maxAttempts = 3
+
+	backoffs := [...]time.Duration{
+		100 * time.Millisecond,
+		500 * time.Millisecond,
+		2500 * time.Millisecond,
+	}
+
+	var err error
+
+	for attempt := range maxAttempts {
+		err = fn()
+		if err == nil {
+			return nil
+		}
+
+		if !isSQLiteTransient(err) {
+			return err
+		}
+
+		if attempt < maxAttempts-1 {
+			s.log.WithError(err).WithField("attempt", attempt+1).
+				Warn("Transient SQLite error, retrying")
+			time.Sleep(backoffs[attempt])
+		}
+	}
+
+	return err
+}
+
+// isSQLiteTransient returns true for transient SQLite errors that may
+// succeed on retry.
+func isSQLiteTransient(err error) bool {
+	msg := err.Error()
+
+	return strings.Contains(msg, "database is locked") ||
+		strings.Contains(msg, "disk I/O error")
+}
+
 // scanMaps scans query results into []map[string]any so only the selected
 // columns appear in the JSON response.
 func scanMaps(

pkg/api/store/store.go

@@ -118,6 +118,17 @@ func (s *store) Start(ctx context.Context) error {
 		return fmt.Errorf("opening database: %w", err)
 	}
 
+	// SQLite requires a single connection to avoid write contention and
+	// ensure pragmas are applied consistently.
+	if s.cfg.Driver == "sqlite" {
+		sqlDB, dbErr := s.db.DB()
+		if dbErr != nil {
+			return fmt.Errorf("getting underlying sql.DB: %w", dbErr)
+		}
+
+		sqlDB.SetMaxOpenConns(1)
+	}
+
 	if err := s.db.WithContext(ctx).AutoMigrate(
 		&User{},
 		&Session{},