perf(storage): allow a small SQLite connection pool

buggregator.yaml.example

@@ -15,6 +15,8 @@ server:
 database:
   driver: sqlite                        # Only SQLite supported for now
   dsn: ${DATABASE_DSN::memory:}         # ":memory:" (default), "data.db", or full path
+  max_open_conns: ${DATABASE_MAX_OPEN_CONNS:4}  # SQLite connection pool size. 1 serializes the UI with ingestion.
+                                        # Needs journal_mode(WAL) in the DSN to be worth anything.
 
 # Attachment storage (SMTP attachments, HTTP dump files)
 storage:

cmd/buggregator/main.go

@@ -48,7 +48,7 @@ func main() {
 	cfg := app.LoadConfig()
 	cfg.Version = version
 
-	db, err := storage.Open(cfg.DatabaseDSN)
+	db, err := storage.OpenPooled(cfg.DatabaseDSN, cfg.Database.MaxOpenConns)
 	if err != nil {
 		slog.Error("failed to open database", "err", err)
 		os.Exit(1)

internal/app/config.go

@@ -6,6 +6,7 @@ import (
 	"log/slog"
 	"os"
 	"regexp"
+	"strconv"
 	"strings"
 
 	"gopkg.in/yaml.v3"
@@ -83,6 +84,15 @@ type StorageConfig struct {
 type DatabaseConfig struct {
 	Driver string `yaml:"driver"`
 	DSN    string `yaml:"dsn"`
+
+	// MaxOpenConns is the size of the SQLite connection pool (default 4).
+	//
+	// With a single connection ingestion and UI queries share one queue: a heavy
+	// list slows down writes, and writes slow down the list. In WAL mode readers
+	// disturb neither each other nor the writer, so a pool is safe; the writer is
+	// still serialized by SQLite itself, and busy_timeout in the DSN makes a
+	// competing write wait instead of failing.
+	MaxOpenConns int `yaml:"max_open_conns"`
 }
 
 type TCPConfig struct {
@@ -177,6 +187,13 @@ func LoadConfig() Config {
 	cfg.Server.Addr = coalesce(cfg.Server.Addr, os.Getenv("HTTP_ADDR"), fileCfg.Server.Addr, ":8000")
 	cfg.Database.DSN = coalesce(cfg.Database.DSN, os.Getenv("DATABASE_DSN"), fileCfg.Database.DSN, ":memory:")
 	cfg.Database.Driver = coalesce(fileCfg.Database.Driver, "sqlite")
+	cfg.Database.MaxOpenConns = 4
+	if v := fileCfg.Database.MaxOpenConns; v > 0 {
+		cfg.Database.MaxOpenConns = v
+	}
+	if v, err := strconv.Atoi(strings.TrimSpace(os.Getenv("DATABASE_MAX_OPEN_CONNS"))); err == nil && v > 0 {
+		cfg.Database.MaxOpenConns = v
+	}
 	cfg.TCP.SMTP.Addr = coalesce(cfg.TCP.SMTP.Addr, os.Getenv("SMTP_ADDR"), fileCfg.TCP.SMTP.Addr, ":1025")
 	cfg.TCP.Monolog.Addr = coalesce(cfg.TCP.Monolog.Addr, os.Getenv("MONOLOG_ADDR"), fileCfg.TCP.Monolog.Addr, ":9913")
 	cfg.TCP.VarDumper.Addr = coalesce(cfg.TCP.VarDumper.Addr, os.Getenv("VAR_DUMPER_ADDR"), fileCfg.TCP.VarDumper.Addr, ":9912")

internal/storage/sqlite.go

@@ -5,6 +5,7 @@ import (
 	"database/sql"
 	"encoding/json"
 	"fmt"
+	"log/slog"
 	"strings"
 
 	"github.com/buggregator/go-buggregator/internal/event"
@@ -16,16 +17,69 @@ type SQLiteStore struct {
 	db *sql.DB
 }
 
-// Open creates a new SQLite database connection.
+// Open creates a new SQLite database connection with a single connection.
 func Open(dsn string) (*sql.DB, error) {
+	return OpenPooled(dsn, 1)
+}
+
+// OpenPooled opens the database with a pool of maxOpen connections.
+//
+// A single connection is only required for writing: in WAL mode readers block
+// neither the writer nor each other, while one shared connection turns every
+// heavy SELECT into a queue for ingestion — the UI and the incoming events wait
+// for each other.
+//
+// There is one correctness requirement: WAL still allows a single writer, so
+// with maxOpen > 1 the DSN needs busy_timeout, otherwise a second write fails
+// with SQLITE_BUSY instead of waiting. Both that and _txlock=immediate are
+// added when absent.
+func OpenPooled(dsn string, maxOpen int) (*sql.DB, error) {
 	if dsn == ":memory:" {
 		dsn = "file::memory:?cache=shared&_pragma=journal_mode(WAL)"
 	}
+	if maxOpen < 1 {
+		maxOpen = 1
+	}
+	if maxOpen > 1 {
+		add := func(param, msg string) {
+			sep := "?"
+			if strings.Contains(dsn, "?") {
+				sep = "&"
+			}
+			if !strings.HasPrefix(dsn, "file:") {
+				dsn = "file:" + dsn
+			}
+			dsn += sep + param
+			slog.Info(msg)
+		}
+
+		if !strings.Contains(dsn, "busy_timeout") {
+			add("_pragma=busy_timeout(10000)",
+				"storage: busy_timeout missing from the DSN, added — a connection pool would otherwise get SQLITE_BUSY")
+		}
+
+		// BEGIN IMMEDIATE instead of BEGIN DEFERRED for every transaction.
+		//
+		// Without it busy_timeout does not help writes: a deferred transaction
+		// starts as a reader and tries to upgrade the lock on its first write —
+		// and if a writer already holds it, SQLite returns SQLITE_BUSY at once
+		// without waiting out the timeout (waiting would deadlock two readers
+		// that both want to write). Verified on a live stream: a pool of four
+		// connections started logging "upsert sentry_traces: database is locked
+		// (5)". Every transaction in the code base is a writing one, so taking
+		// the write lock upfront costs nothing.
+		if !strings.Contains(dsn, "_txlock") {
+			add("_txlock=immediate",
+				"storage: _txlock missing from the DSN, added immediate — a pooled write would otherwise get SQLITE_BUSY without waiting")
+		}
+	}
+
 	db, err := sql.Open("sqlite", dsn)
 	if err != nil {
 		return nil, err
 	}
-	db.SetMaxOpenConns(1) // SQLite single-writer
+	db.SetMaxOpenConns(maxOpen)
+	db.SetMaxIdleConns(maxOpen)
 	return db, nil
 }
 

internal/storage/sqlite_test.go

@@ -4,6 +4,9 @@ import (
 	"context"
 	"database/sql"
 	"encoding/json"
+	"path/filepath"
+	"strconv"
+	"sync"
 	"testing"
 
 	"github.com/buggregator/go-buggregator/internal/event"
@@ -245,3 +248,80 @@ func TestSQLiteStore_Pin_Unpin(t *testing.T) {
 		t.Error("expected IsPinned to be false")
 	}
 }
+
+// A pool of connections must not turn concurrent writes into SQLITE_BUSY. Two
+// things make that work and both are added to the DSN by OpenPooled: a
+// busy_timeout, so a competing write waits, and _txlock=immediate, so a
+// transaction takes the write lock upfront instead of failing on the upgrade
+// (a deferred transaction that starts as a reader gets SQLITE_BUSY immediately,
+// without waiting out the timeout).
+func TestOpenPooled_ConcurrentWritesAndReads(t *testing.T) {
+	dsn := "file:" + filepath.Join(t.TempDir(), "pool.db") + "?_pragma=journal_mode(WAL)"
+	db, err := storage.OpenPooled(dsn, 4)
+	if err != nil {
+		t.Fatal(err)
+	}
+	defer db.Close()
+
+	if _, err := db.Exec(`CREATE TABLE events (
+		uuid TEXT PRIMARY KEY, type TEXT NOT NULL, payload TEXT NOT NULL,
+		timestamp TEXT NOT NULL, project TEXT, is_pinned INTEGER NOT NULL DEFAULT 0
+	)`); err != nil {
+		t.Fatal(err)
+	}
+
+	store := storage.NewSQLiteStore(db)
+	ctx := context.Background()
+
+	const writers, perWriter = 4, 25
+	errs := make(chan error, writers*perWriter)
+	var wg sync.WaitGroup
+
+	for w := 0; w < writers; w++ {
+		wg.Add(1)
+		go func(w int) {
+			defer wg.Done()
+			for i := 0; i < perWriter; i++ {
+				uuid := "w" + strconv.Itoa(w) + "-" + strconv.Itoa(i)
+				if err := store.Store(ctx, makeEvent(uuid, "sentry", "default")); err != nil {
+					errs <- err
+					return
+				}
+				// A reader running against the same pool while writes are in flight.
+				if _, err := store.FindAll(ctx, event.FindOptions{Project: "default", Limit: 10}); err != nil {
+					errs <- err
+					return
+				}
+			}
+		}(w)
+	}
+	wg.Wait()
+	close(errs)
+
+	for err := range errs {
+		t.Fatalf("concurrent access failed: %v", err)
+	}
+
+	var count int
+	if err := db.QueryRow(`SELECT COUNT(*) FROM events`).Scan(&count); err != nil {
+		t.Fatal(err)
+	}
+	if count != writers*perWriter {
+		t.Fatalf("stored %d events, want %d", count, writers*perWriter)
+	}
+}
+
+// A caller that asks for one connection (or a nonsensical number) keeps the
+// historical behaviour and an untouched DSN.
+func TestOpenPooled_SingleConnection(t *testing.T) {
+	for _, maxOpen := range []int{0, 1} {
+		db, err := storage.OpenPooled(":memory:", maxOpen)
+		if err != nil {
+			t.Fatalf("maxOpen=%d: %v", maxOpen, err)
+		}
+		if got := db.Stats().MaxOpenConnections; got != 1 {
+			t.Errorf("maxOpen=%d: MaxOpenConnections = %d, want 1", maxOpen, got)
+		}
+		db.Close()
+	}
+}