fix: database locks and busy

Author: ItsNotGoodName

Makefile

@@ -50,7 +50,7 @@ db-migrate:
 # Generation
 
 gen-jet:
-	jet -source=sqlite -dsn="$(DB_PATH)" -path=./internal/jet -ignore-tables goose_db_version
+	jet -source=sqlite -dsn="$(DB_PATH)" -path=./internal/jet -ignore-tables goose_db_version,_dummy
 	rm -rf ./internal/jet/model
 
 gen-templ:

cmd/stress-database/main.go

@@ -0,0 +1,53 @@
+package main
+
+import (
+	"context"
+	"errors"
+	"fmt"
+	"log"
+	"os"
+	"os/signal"
+	"time"
+
+	"github.com/ItsNotGoodName/smtpbridge/internal/database"
+	"github.com/ItsNotGoodName/smtpbridge/internal/models"
+	"github.com/ItsNotGoodName/smtpbridge/internal/repo"
+)
+
+func main() {
+	ctx, cancel := signal.NotifyContext(context.Background(), os.Interrupt)
+	defer cancel()
+
+	db, err := database.New("./smtpbridge_data/smtpbridge.db", false)
+	if err != nil {
+		log.Fatalln(err)
+	}
+
+	go func() {
+		msg := models.Message{}
+		for {
+			id, err := repo.EnvelopeCreate(ctx, db, msg, []models.Attachment{})
+			if err != nil {
+				log.Fatalln(err)
+			}
+
+			err = repo.MailmanEnqueue(ctx, db, id)
+			if err != nil {
+				log.Fatalln(err)
+			}
+		}
+	}()
+
+	for {
+		id, err := repo.MailmanDequeue(ctx, db)
+		if err != nil {
+			if !errors.Is(err, repo.ErrNoRows) {
+				log.Fatalln(err)
+			}
+
+			time.Sleep(1 * time.Second)
+		}
+
+		fmt.Println(id)
+	}
+}

internal/database/database.go

@@ -12,7 +12,7 @@ type Querier interface {
 	Conn() *sql.DB
 	ExecContext(ctx context.Context, query string, args ...any) (sql.Result, error)
 	QueryContext(ctx context.Context, query string, args ...any) (*sql.Rows, error)
-	BeginTx(ctx context.Context, opts *sql.TxOptions) (QuerierTx, error)
+	BeginTx(ctx context.Context, write bool) (QuerierTx, error)
 }
 
 type QuerierTx interface {
@@ -34,6 +34,10 @@ func New(dbPath string, debug bool) (Querier, error) {
 		return nil, err
 	}
 
+	if err := dummyCreate(db); err != nil {
+		return nil, err
+	}
+
 	if debug {
 		return DebugDB{DB: db}, nil
 	}
@@ -45,8 +49,8 @@ type DB struct {
 	*sql.DB
 }
 
-func (db DB) BeginTx(ctx context.Context, opts *sql.TxOptions) (QuerierTx, error) {
-	return db.DB.BeginTx(ctx, opts)
+func (db DB) BeginTx(ctx context.Context, write bool) (QuerierTx, error) {
+	return dummyBeginTx(ctx, db.DB, write)
 }
 
 func (db DB) Conn() *sql.DB {
@@ -97,20 +101,16 @@ func (tx DebugTx) QueryContext(ctx context.Context, query string, args ...any) (
 	return tx.Tx.QueryContext(ctx, query, args...)
 }
 
-func (db DebugDB) BeginTx(ctx context.Context, opts *sql.TxOptions) (QuerierTx, error) {
+func (db DebugDB) BeginTx(ctx context.Context, write bool) (QuerierTx, error) {
 	log.Debug().
 		Msg("BeginTx (Tx)")
-	tx, err := db.DB.BeginTx(ctx, opts)
+	tx, err := dummyBeginTx(ctx, db.DB, write)
 	if err != nil {
 		return DebugTx{}, err
 	}
 	return DebugTx{Tx: tx}, nil
 }
 
-type DebugTx struct {
-	*sql.Tx
-}
-
 func (tx DebugTx) Commit() error {
 	log.Debug().
 		Str("func", "Commit (Tx)").
@@ -124,3 +124,7 @@ func (tx DebugTx) Rollback() error {
 		Msg("")
 	return tx.Tx.Rollback()
 }
+
+type DebugTx struct {
+	*sql.Tx
+}

internal/database/dummy.go

@@ -0,0 +1,29 @@
+package database
+
+import (
+	"context"
+	"database/sql"
+)
+
+func dummyCreate(db *sql.DB) error {
+	_, err := db.Exec("CREATE TABLE IF NOT EXISTS _dummy(_dummy integer);")
+	return err
+}
+
+func dummyBeginTx(ctx context.Context, db *sql.DB, write bool) (*sql.Tx, error) {
+	tx, err := db.BeginTx(ctx, nil)
+	if err != nil {
+		return nil, err
+	}
+
+	if write {
+		// This prevents SQLITE_BUSY (5) and database locked (517) when doing write transactions
+		// because we tell sqlite that we are going to do a write transaction through the dummy DELETE query.
+		_, err = tx.ExecContext(ctx, "DELETE FROM _dummy WHERE 0 = 1;")
+		if err != nil {
+			return nil, err
+		}
+	}
+
+	return tx, nil
+}

internal/repo/envelope.go

@@ -31,7 +31,7 @@ var attachmentPJ ProjectionList = ProjectionList{
 }
 
 func EnvelopeCreate(ctx context.Context, db database.Querier, msg models.Message, atts []models.Attachment) (int64, error) {
-	tx, err := db.BeginTx(ctx, nil)
+	tx, err := db.BeginTx(ctx, true)
 	if err != nil {
 		return 0, err
 	}

internal/repo/internal.go

@@ -20,7 +20,7 @@ func InternalSync(
 	rules []models.Rule,
 	ruleToEndpoints map[string][]string,
 ) error {
-	tx, err := db.BeginTx(ctx, nil)
+	tx, err := db.BeginTx(ctx, true)
 	if err != nil {
 		return err
 	}

internal/repo/mailman.go

@@ -25,7 +25,7 @@ func MailmanEnqueue(ctx context.Context, db database.Querier, envelopeID int64)
 }
 
 func MailmanDequeue(ctx context.Context, db database.Querier) (int64, error) {
-	tx, err := db.BeginTx(ctx, nil)
+	tx, err := db.BeginTx(ctx, true)
 	if err != nil {
 		return 0, err
 	}

internal/repo/rule.go

@@ -22,7 +22,7 @@ var rulePJ ProjectionList = ProjectionList{
 var ruleCreatePJ ProjectionList = rulePJ.Except(Rules.ID)
 
 func RuleCreate(ctx context.Context, db database.Querier, rule models.Rule, endpoints []int64) (int64, error) {
-	tx, err := db.BeginTx(ctx, nil)
+	tx, err := db.BeginTx(ctx, true)
 	if err != nil {
 		return 0, err
 	}
@@ -129,7 +129,7 @@ func RuleEndpointsList(ctx context.Context, db database.Querier) ([]models.RuleE
 }
 
 func RuleEndpointsSet(ctx context.Context, db database.Querier, ruleID int64, endpointIDs []int64) error {
-	tx, err := db.BeginTx(ctx, nil)
+	tx, err := db.BeginTx(ctx, true)
 	if err != nil {
 		return err
 	}