diff --git a/buggregator.yaml.example b/buggregator.yaml.example index a2bbbc4..609abc8 100644 --- a/buggregator.yaml.example +++ b/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: diff --git a/cmd/buggregator/main.go b/cmd/buggregator/main.go index b7e7723..3dbe5dd 100644 --- a/cmd/buggregator/main.go +++ b/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) diff --git a/internal/app/config.go b/internal/app/config.go index b482656..631bb25 100644 --- a/internal/app/config.go +++ b/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") diff --git a/internal/storage/sqlite.go b/internal/storage/sqlite.go index 6c13dda..056bcef 100644 --- a/internal/storage/sqlite.go +++ b/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 } diff --git a/internal/storage/sqlite_test.go b/internal/storage/sqlite_test.go index 1722aca..a0e2008 100644 --- a/internal/storage/sqlite_test.go +++ b/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() + } +}