修复SQLite连接配置与截止时间锁等待并补充并发回归

This commit is contained in:
xuans 2026-09-28 01:06:01 +08:00
parent 6cab85d532
commit af831aeea3
11 changed files with 1064 additions and 24 deletions

View File

@ -0,0 +1,58 @@
package cli
import (
"database/sql"
"os"
"path/filepath"
"testing"
"git.pcux.cn/xuans/Scert/backend/internal/config"
"git.pcux.cn/xuans/Scert/backend/internal/database"
)
func TestSQLiteSnapshotWhileWriterIsActive(t *testing.T) {
cfg := config.Default(t.TempDir())
db, err := database.Open(t.Context(), cfg.Database)
if err != nil {
t.Fatal(err)
}
defer db.Close()
if _, err := db.DB.ExecContext(t.Context(), "CREATE TABLE snapshot_probe (id INTEGER PRIMARY KEY, value INTEGER); INSERT INTO snapshot_probe VALUES (1, 7)"); err != nil {
t.Fatal(err)
}
if info, err := os.Stat(cfg.Database.DSN + "-wal"); err != nil || info.Size() == 0 {
t.Fatalf("没有实际生成 WAL 数据:%v", err)
}
writer, err := db.DB.BeginTx(t.Context(), nil)
if err != nil {
t.Fatal(err)
}
defer writer.Rollback()
if _, err := writer.ExecContext(t.Context(), "UPDATE snapshot_probe SET value = 99 WHERE id = 1"); err != nil {
t.Fatal(err)
}
for _, name := range []string{"uncommitted", "committed"} {
t.Run(name, func(t *testing.T) {
want := 7
if name == "committed" {
if err := writer.Commit(); err != nil {
t.Fatal(err)
}
want = 99
}
output := filepath.Join(t.TempDir(), "snapshot.db")
if err := backupSQLiteDatabase(t.Context(), cfg, output); err != nil {
t.Fatalf("活动文件库快照失败:%v", err)
}
snapshot, err := sql.Open("sqlite", output)
if err != nil {
t.Fatal(err)
}
defer snapshot.Close()
var value int
if err := snapshot.QueryRowContext(t.Context(), "SELECT value FROM snapshot_probe WHERE id = 1").Scan(&value); err != nil || value != want {
t.Fatalf("快照没有保存对应提交状态:得到 %d,期望 %d,错误=%v", value, want, err)
}
})
}
}

View File

@ -12,7 +12,6 @@ import (
"github.com/go-sql-driver/mysql"
"github.com/jackc/pgx/v5"
_ "github.com/jackc/pgx/v5/stdlib"
_ "modernc.org/sqlite"
"git.pcux.cn/xuans/Scert/backend/internal/config"
)
@ -59,6 +58,8 @@ func Open(ctx context.Context, cfg config.DatabaseConfig) (*Database, error) {
var err error
if driver == "mysql" {
db, err = openMySQLPool(dsn)
} else if driver == "sqlite" {
db = openSQLitePool(dsn)
} else {
db, err = sql.Open(sqlDriver, dsn)
}
@ -74,12 +75,6 @@ func Open(ctx context.Context, cfg config.DatabaseConfig) (*Database, error) {
_ = db.Close()
return nil, fmt.Errorf("连接数据库失败:%w", err)
}
if driver == "sqlite" {
if err := configureSQLite(pingCtx, db); err != nil {
_ = db.Close()
return nil, err
}
}
return &Database{DB: db}, nil
}
@ -99,23 +94,6 @@ func ValidateConfigForOpen(cfg config.DatabaseConfig) error {
return nil
}
func configureSQLite(ctx context.Context, db *sql.DB) error {
if _, err := db.ExecContext(ctx, "PRAGMA busy_timeout = 5000"); err != nil {
return fmt.Errorf("配置 SQLite 忙等待失败:%w", err)
}
var mode string
if _, err := db.ExecContext(ctx, "PRAGMA journal_mode = WAL"); err != nil {
return fmt.Errorf("配置 SQLite WAL 失败:%w", err)
}
if err := db.QueryRowContext(ctx, "PRAGMA journal_mode").Scan(&mode); err != nil {
return fmt.Errorf("读取 SQLite 日志模式失败:%w", err)
}
if strings.EqualFold(strings.TrimSpace(mode), "memory") || strings.EqualFold(strings.TrimSpace(mode), "wal") {
return nil
}
return fmt.Errorf("SQLite 日志模式无效:%s", mode)
}
func normalizeMySQLDSN(dsn string) (string, error) {
cfg, err := mysql.ParseDSN(dsn)
if err != nil {

View File

@ -0,0 +1,208 @@
package database
import (
"context"
"database/sql"
"errors"
"fmt"
"sync"
"testing"
"time"
"git.pcux.cn/xuans/Scert/backend/internal/config"
)
func newSQLiteConcurrencyPools(t *testing.T) (*sql.DB, *sql.DB) {
t.Helper()
cfg := config.Default(t.TempDir()).Database
first := openSQLiteTestPool(t, cfg)
second := openSQLiteTestPool(t, cfg)
if _, err := first.ExecContext(t.Context(), "CREATE TABLE lock_probe (id INTEGER PRIMARY KEY, value INTEGER NOT NULL); INSERT INTO lock_probe VALUES (1, 0)"); err != nil {
t.Fatalf("创建隔离并发数据失败:%v", err)
}
return first, second
}
func readSQLiteProbe(t *testing.T, db interface {
QueryRowContext(context.Context, string, ...any) *sql.Row
}) int {
t.Helper()
var value int
if err := db.QueryRowContext(t.Context(), "SELECT value FROM lock_probe WHERE id = 1").Scan(&value); err != nil {
t.Fatalf("读取隔离并发数据失败:%v", err)
}
return value
}
func TestSQLiteConcurrentReadersAndWriter(t *testing.T) {
first, second := newSQLiteConcurrencyPools(t)
reader, err := first.BeginTx(t.Context(), nil)
if err != nil {
t.Fatal(err)
}
defer reader.Rollback()
if value := readSQLiteProbe(t, reader); value != 0 {
t.Fatalf("初始快照错误:%d", value)
}
writer, err := second.BeginTx(t.Context(), nil)
if err != nil {
t.Fatal(err)
}
defer writer.Rollback()
if _, err := writer.ExecContext(t.Context(), "UPDATE lock_probe SET value = 1 WHERE id = 1"); err != nil {
t.Fatalf("读事务存在时写入失败:%v", err)
}
if value := readSQLiteProbe(t, first); value != 0 {
t.Fatalf("其他连接读到了未提交数据:%d", value)
}
if err := writer.Commit(); err != nil {
t.Fatalf("读事务存在时提交失败:%v", err)
}
if value := readSQLiteProbe(t, reader); value != 0 {
t.Fatalf("已有读事务的快照发生漂移:%d", value)
}
if value := readSQLiteProbe(t, first); value != 1 {
t.Fatalf("新查询没有读到提交数据:%d", value)
}
}
func TestSQLiteWriterWaitsForOtherPool(t *testing.T) {
first, second := newSQLiteConcurrencyPools(t)
writer, err := first.BeginTx(t.Context(), nil)
if err != nil {
t.Fatal(err)
}
defer writer.Rollback()
if _, err := writer.ExecContext(t.Context(), "UPDATE lock_probe SET value = 1 WHERE id = 1"); err != nil {
t.Fatal(err)
}
ctx, cancel := context.WithTimeout(t.Context(), 8*time.Second)
defer cancel()
started := make(chan struct{})
done := make(chan error, 1)
go func() {
close(started)
_, err := second.ExecContext(ctx, "UPDATE lock_probe SET value = value + 1 WHERE id = 1")
done <- err
}()
<-started
select {
case err := <-done:
t.Fatalf("持锁期间竞争写入没有等待:%v", err)
case <-time.After(150 * time.Millisecond):
}
if err := writer.Commit(); err != nil {
t.Fatal(err)
}
select {
case err := <-done:
if err != nil {
t.Fatalf("释放写锁后等待方未成功写入:%v", err)
}
case <-ctx.Done():
t.Fatalf("释放写锁后等待方没有结束:%v", ctx.Err())
}
if value := readSQLiteProbe(t, second); value != 2 {
t.Fatalf("两次提交结果不正确:%d", value)
}
}
func TestSQLiteConcurrentWritesAcrossPools(t *testing.T) {
first, second := newSQLiteConcurrencyPools(t)
ctx, cancel := context.WithTimeout(t.Context(), 20*time.Second)
defer cancel()
const workers, writes = 8, 25
start := make(chan struct{})
results := make(chan error, workers)
var wg sync.WaitGroup
for i := 0; i < workers; i++ {
db := []*sql.DB{first, second}[i%2]
wg.Go(func() {
<-start
for j := 0; j < writes; j++ {
if _, err := db.ExecContext(ctx, "UPDATE lock_probe SET value = value + 1 WHERE id = 1"); err != nil {
results <- fmt.Errorf("第 %d 次并发写入失败:%w", j+1, err)
return
}
}
})
}
close(start)
wg.Wait()
close(results)
for err := range results {
t.Error(err)
}
if value := readSQLiteProbe(t, first); value != workers*writes {
t.Fatalf("并发写入丢失:得到 %d,期望 %d", value, workers*writes)
}
if first.Stats().InUse != 0 || second.Stats().InUse != 0 {
t.Fatal("并发操作结束后连接未释放")
}
}
func TestSQLiteTransactionRollbackAndCancellation(t *testing.T) {
for _, name := range []string{"rollback", "cancel"} {
t.Run(name, func(t *testing.T) {
first, second := newSQLiteConcurrencyPools(t)
ctx, cancel := context.WithCancel(t.Context())
defer cancel()
tx, err := first.BeginTx(ctx, nil)
if err != nil {
t.Fatal(err)
}
defer tx.Rollback()
if _, err := tx.ExecContext(ctx, "UPDATE lock_probe SET value = 9 WHERE id = 1"); err != nil {
t.Fatal(err)
}
if name == "cancel" {
cancel()
if err := tx.Commit(); !errors.Is(err, context.Canceled) && !errors.Is(err, sql.ErrTxDone) {
t.Fatalf("取消后的事务没有拒绝提交:%v", err)
}
} else if err := tx.Rollback(); err != nil {
t.Fatal(err)
}
// 独立连接池能够写入,证明回滚实际释放了锁,而非只返回错误。
if _, err := second.ExecContext(t.Context(), "UPDATE lock_probe SET value = value + 1 WHERE id = 1"); err != nil {
t.Fatalf("事务结束后写锁未释放:%v", err)
}
if value := readSQLiteProbe(t, second); value != 1 {
t.Fatalf("被取消或回滚的数据仍然存在:%d", value)
}
})
}
}
func TestSQLiteLockWaitCancellation(t *testing.T) {
first, second := newSQLiteConcurrencyPools(t)
writer, err := first.BeginTx(t.Context(), nil)
if err != nil {
t.Fatal(err)
}
defer writer.Rollback()
if _, err := writer.ExecContext(t.Context(), "UPDATE lock_probe SET value = 1 WHERE id = 1"); err != nil {
t.Fatal(err)
}
ctx, cancel := context.WithTimeout(t.Context(), 150*time.Millisecond)
defer cancel()
started := time.Now()
_, err = second.ExecContext(ctx, "UPDATE lock_probe SET value = 99 WHERE id = 1")
elapsed := time.Since(started)
t.Logf("写锁等待取消耗时:%s", elapsed)
if !errors.Is(err, context.DeadlineExceeded) {
t.Errorf("等待中的写入没有返回超时错误:%v", err)
}
if elapsed > time.Second {
t.Errorf("取消响应超过 1 秒:%s", elapsed)
}
if err := writer.Rollback(); err != nil {
t.Fatal(err)
}
if value := readSQLiteProbe(t, second); value != 0 {
t.Fatalf("被取消的写入产生了持久化副作用:%d", value)
}
if _, err := second.ExecContext(t.Context(), "UPDATE lock_probe SET value = 2 WHERE id = 1"); err != nil {
t.Fatalf("取消后连接不能继续使用:%v", err)
}
}

View File

@ -0,0 +1,154 @@
package database
import (
"context"
"database/sql/driver"
"errors"
"fmt"
"time"
"modernc.org/sqlite"
)
// 保留锁定驱动的原有接口,不通过重放业务语句实现锁等待或超时。
type sqliteDriverConn interface {
driver.Conn
driver.ConnBeginTx
driver.ConnPrepareContext
driver.ExecerContext
driver.QueryerContext
driver.Execer
driver.Queryer
driver.Pinger
driver.SessionResetter
driver.Validator
Serialize() ([]byte, error)
Deserialize([]byte) error
NewBackup(string) (*sqlite.Backup, error)
NewRestore(string) (*sqlite.Backup, error)
}
type sqliteDeadlineConn struct {
sqliteDriverConn
invalid bool
}
func (c *sqliteDeadlineConn) IsValid() bool {
return !c.invalid && c.sqliteDriverConn.IsValid()
}
func (c *sqliteDeadlineConn) ResetSession(ctx context.Context) error {
if c.invalid {
return driver.ErrBadConn
}
return c.sqliteDriverConn.ResetSession(ctx)
}
// SQLite 原生忙等待可能晚于上下文截止时间;缩短本次调用的等待预算,结束后恢复。
// 不修改没有截止时间的调用;手动取消仍依赖原生驱动中断,不能承诺立即退出忙等待。
func sqliteWithDeadline[T any](c *sqliteDeadlineConn, ctx context.Context, run func() (T, error)) (result T, err error) {
if err := ctx.Err(); err != nil {
return result, err
}
defer func() {
var sqliteErr *sqlite.Error
if ctx.Err() != nil && errors.As(err, &sqliteErr) && (sqliteErr.Code()&255 == 5 || sqliteErr.Code()&255 == 9) {
// 原生 BEGIN 不一定转换上下文错误;保留 SQLite 错误供调用者诊断。
err = errors.Join(ctx.Err(), err)
}
}()
deadline, limited := ctx.Deadline()
remaining := time.Until(deadline)
if !limited || remaining >= 5*time.Second {
return run()
}
if remaining <= 0 {
return result, context.DeadlineExceeded
}
// 向上取整,避免在截止时间之前以 SQLITE_BUSY 提前返回。
millis := (remaining + time.Millisecond - 1) / time.Millisecond
if _, err := c.sqliteDriverConn.ExecContext(ctx, fmt.Sprintf("PRAGMA busy_timeout = %d", millis), nil); err != nil {
c.invalid = true
return result, fmt.Errorf("配置 SQLite 截止时间等待预算失败:%w", err)
}
defer func() {
if _, restoreErr := c.sqliteDriverConn.ExecContext(context.Background(), "PRAGMA busy_timeout = 5000", nil); restoreErr != nil {
c.invalid = true
// 执行后的清理错误不能转换成 ErrBadConn,避免数据库层自动重试已执行语句。
err = errors.Join(err, fmt.Errorf("恢复 SQLite 忙等待配置失败:%w", restoreErr))
}
}()
result, err = run()
return result, err
}
func (c *sqliteDeadlineConn) ExecContext(ctx context.Context, query string, args []driver.NamedValue) (driver.Result, error) {
return sqliteWithDeadline(c, ctx, func() (driver.Result, error) {
return c.sqliteDriverConn.ExecContext(ctx, query, args)
})
}
func (c *sqliteDeadlineConn) Ping(ctx context.Context) error {
_, err := c.ExecContext(ctx, "SELECT 1", nil)
return err
}
func (c *sqliteDeadlineConn) QueryContext(ctx context.Context, query string, args []driver.NamedValue) (driver.Rows, error) {
rows, err := sqliteWithDeadline(c, ctx, func() (driver.Rows, error) {
return c.sqliteDriverConn.QueryContext(ctx, query, args)
})
return c.wrapRows(ctx, rows, err)
}
func (c *sqliteDeadlineConn) BeginTx(ctx context.Context, options driver.TxOptions) (driver.Tx, error) {
return sqliteWithDeadline(c, ctx, func() (driver.Tx, error) {
return c.sqliteDriverConn.BeginTx(ctx, options)
})
}
func (c *sqliteDeadlineConn) PrepareContext(ctx context.Context, query string) (driver.Stmt, error) {
if err := ctx.Err(); err != nil {
return nil, err
}
stmt, err := c.sqliteDriverConn.PrepareContext(ctx, query)
if err != nil {
return nil, err
}
full, ok := stmt.(sqliteDriverStmt)
if !ok {
_ = stmt.Close()
return nil, fmt.Errorf("SQLite 驱动预处理接口不兼容")
}
return &sqliteDeadlineStmt{sqliteDriverStmt: full, conn: c}, nil
}
func (c *sqliteDeadlineConn) Prepare(query string) (driver.Stmt, error) {
return c.PrepareContext(context.Background(), query)
}
type sqliteDriverStmt interface {
driver.Stmt
driver.StmtExecContext
driver.StmtQueryContext
}
type sqliteDeadlineStmt struct {
sqliteDriverStmt
conn *sqliteDeadlineConn
}
func (s *sqliteDeadlineStmt) ExecContext(ctx context.Context, args []driver.NamedValue) (driver.Result, error) {
return sqliteWithDeadline(s.conn, ctx, func() (driver.Result, error) {
return s.sqliteDriverStmt.ExecContext(ctx, args)
})
}
func (s *sqliteDeadlineStmt) QueryContext(ctx context.Context, args []driver.NamedValue) (driver.Rows, error) {
rows, err := sqliteWithDeadline(s.conn, ctx, func() (driver.Rows, error) {
return s.sqliteDriverStmt.QueryContext(ctx, args)
})
return s.conn.wrapRows(ctx, rows, err)
}
var _ sqliteDriverConn = (*sqliteDeadlineConn)(nil)
var _ sqliteDriverStmt = (*sqliteDeadlineStmt)(nil)

View File

@ -0,0 +1,79 @@
package database
import (
"context"
"database/sql/driver"
"errors"
"strings"
"testing"
"time"
)
type sqliteDeadlineFaultConn struct {
sqliteDriverConn
commands []string
failAt int
err error
}
func (c *sqliteDeadlineFaultConn) ExecContext(ctx context.Context, query string, args []driver.NamedValue) (driver.Result, error) {
c.commands = append(c.commands, query)
if len(c.commands) == c.failAt {
return nil, c.err
}
return driver.RowsAffected(0), nil
}
func (c *sqliteDeadlineFaultConn) IsValid() bool { return true }
func TestSQLiteDeadlineFailureNeverReplaysOperation(t *testing.T) {
for _, name := range []string{"setup_failure", "restore_failure", "operation_failure", "already_canceled"} {
t.Run(name, func(t *testing.T) {
failure := errors.New("注入的隔离测试错误")
raw := &sqliteDeadlineFaultConn{err: failure}
conn := &sqliteDeadlineConn{sqliteDriverConn: raw}
ctx, cancel := context.WithTimeout(t.Context(), time.Second)
defer cancel()
wantRuns := 1
switch name {
case "setup_failure":
raw.failAt = 1
wantRuns = 0
case "restore_failure":
raw.failAt = 2
case "already_canceled":
cancel()
wantRuns = 0
}
runs := 0
_, err := sqliteWithDeadline(conn, ctx, func() (int, error) {
runs++
if name == "operation_failure" {
return 0, failure
}
return 7, nil
})
if err == nil || runs != wantRuns || errors.Is(err, driver.ErrBadConn) {
t.Fatalf("错误处理发生重放或丢失:执行次数=%d,错误=%v", runs, err)
}
if name == "setup_failure" || name == "restore_failure" {
if conn.IsValid() || !errors.Is(conn.ResetSession(t.Context()), driver.ErrBadConn) {
t.Fatal("配置状态不确定的连接仍可被复用")
}
}
if name == "operation_failure" {
if !errors.Is(err, failure) || len(raw.commands) != 2 || raw.commands[1] != "PRAGMA busy_timeout = 5000" || !conn.IsValid() {
t.Fatalf("业务错误后没有保留错误或恢复连接配置:%v", raw.commands)
}
}
if name == "already_canceled" && (len(raw.commands) != 0 || !errors.Is(err, context.Canceled)) {
t.Fatal("已经取消的调用仍然修改了连接")
}
for _, command := range raw.commands {
if !strings.HasPrefix(command, "PRAGMA busy_timeout = ") {
t.Fatalf("执行了额外 SQL:%s", command)
}
}
})
}
}

View File

@ -0,0 +1,46 @@
package database
import (
"context"
"database/sql/driver"
"fmt"
)
type sqliteDriverRows interface {
driver.Rows
driver.RowsColumnTypeDatabaseTypeName
driver.RowsColumnTypeLength
driver.RowsColumnTypeNullable
driver.RowsColumnTypePrecisionScale
driver.RowsColumnTypeScanType
}
type sqliteDeadlineRows struct {
sqliteDriverRows
conn *sqliteDeadlineConn
ctx context.Context
}
func (c *sqliteDeadlineConn) wrapRows(ctx context.Context, rows driver.Rows, err error) (driver.Rows, error) {
if err != nil {
if rows != nil {
_ = rows.Close()
}
return nil, err
}
full, ok := rows.(sqliteDriverRows)
if !ok {
_ = rows.Close()
return nil, fmt.Errorf("SQLite 驱动查询结果接口不兼容")
}
return &sqliteDeadlineRows{sqliteDriverRows: full, conn: c, ctx: ctx}, nil
}
func (r *sqliteDeadlineRows) Next(values []driver.Value) error {
_, err := sqliteWithDeadline(r.conn, r.ctx, func() (struct{}, error) {
return struct{}{}, r.sqliteDriverRows.Next(values)
})
return err
}
var _ sqliteDriverRows = (*sqliteDeadlineRows)(nil)

View File

@ -0,0 +1,186 @@
package database
import (
"context"
"database/sql"
"errors"
"testing"
"time"
"git.pcux.cn/xuans/Scert/backend/internal/config"
)
func TestSQLiteDeadlineAcrossExecutionPaths(t *testing.T) {
for _, name := range []string{"exec", "query", "prepared_exec", "prepared_query", "transaction", "immediate_transaction"} {
t.Run(name, func(t *testing.T) {
first, second := newSQLiteConcurrencyPools(t)
if name == "immediate_transaction" {
var filename string
var sequence int
var databaseName string
if err := first.QueryRowContext(t.Context(), "PRAGMA database_list").Scan(&sequence, &databaseName, &filename); err != nil {
t.Fatal(err)
}
cfg := config.Default(t.TempDir()).Database
cfg.DSN = filename + "?_txlock=immediate"
second = openSQLiteTestPool(t, cfg)
}
conn, err := second.Conn(t.Context())
if err != nil {
t.Fatal(err)
}
defer conn.Close()
writer, err := first.BeginTx(t.Context(), nil)
if err != nil {
t.Fatal(err)
}
defer writer.Rollback()
if _, err := writer.ExecContext(t.Context(), "UPDATE lock_probe SET value = 1 WHERE id = 1"); err != nil {
t.Fatal(err)
}
ctx, cancel := context.WithTimeout(t.Context(), 150*time.Millisecond)
defer cancel()
started := time.Now()
switch name {
case "exec":
_, err = conn.ExecContext(ctx, "UPDATE lock_probe SET value = 99 WHERE id = 1")
case "query":
var value int
err = conn.QueryRowContext(ctx, "UPDATE lock_probe SET value = 99 WHERE id = 1 RETURNING value").Scan(&value)
case "prepared_exec", "prepared_query":
statement := "UPDATE lock_probe SET value = ? WHERE id = 1"
if name == "prepared_query" {
statement += " RETURNING value"
}
stmt, prepareErr := conn.PrepareContext(t.Context(), statement)
if prepareErr != nil {
t.Fatal(prepareErr)
}
defer stmt.Close()
if name == "prepared_exec" {
_, err = stmt.ExecContext(ctx, 99)
} else {
var value int
err = stmt.QueryRowContext(ctx, 99).Scan(&value)
}
case "transaction":
tx, beginErr := conn.BeginTx(t.Context(), nil)
if beginErr != nil {
t.Fatal(beginErr)
}
defer tx.Rollback()
_, err = tx.ExecContext(ctx, "UPDATE lock_probe SET value = 99 WHERE id = 1")
_ = tx.Rollback()
case "immediate_transaction":
var tx *sql.Tx
tx, err = conn.BeginTx(ctx, nil)
if tx != nil {
_ = tx.Rollback()
}
}
elapsed := time.Since(started)
if !errors.Is(err, context.DeadlineExceeded) || elapsed > time.Second {
t.Errorf("执行路径未遵守截止时间:耗时=%s,错误=%v", elapsed, err)
}
checkSQLiteConnection(t, t.Context(), conn)
if err := writer.Rollback(); err != nil {
t.Fatal(err)
}
if value := readSQLiteProbe(t, conn); value != 0 {
t.Fatalf("超时写入产生了副作用:%d", value)
}
if _, err := conn.ExecContext(t.Context(), "UPDATE lock_probe SET value = 2 WHERE id = 1"); err != nil {
t.Fatalf("超时后的同一连接无法继续写入:%v", err)
}
})
}
}
func TestSQLitePoolWaitHonorsContext(t *testing.T) {
cfg := config.Default(t.TempDir()).Database
cfg.MaxOpenConns = 1
cfg.MaxIdleConns = 1
db := openSQLiteTestPool(t, cfg)
conn, err := db.Conn(t.Context())
if err != nil {
t.Fatal(err)
}
defer conn.Close()
ctx, cancel := context.WithTimeout(t.Context(), 100*time.Millisecond)
defer cancel()
if _, err := db.ExecContext(ctx, "SELECT 1"); !errors.Is(err, context.DeadlineExceeded) {
t.Fatalf("连接池耗尽时没有响应截止时间:%v", err)
}
if err := conn.Close(); err != nil {
t.Fatal(err)
}
if _, err := db.ExecContext(t.Context(), "SELECT 1"); err != nil {
t.Fatalf("等待取消后连接池未恢复:%v", err)
}
}
func TestSQLiteDeadlinePreservesQueryMetadata(t *testing.T) {
first, _ := newSQLiteConcurrencyPools(t)
ctx, cancel := context.WithTimeout(t.Context(), 2*time.Second)
defer cancel()
rows, err := first.QueryContext(ctx, "SELECT id, value FROM lock_probe")
if err != nil {
t.Fatal(err)
}
defer rows.Close()
columns, err := rows.ColumnTypes()
if err != nil || len(columns) != 2 || columns[0].DatabaseTypeName() != "INTEGER" {
t.Fatalf("包装后的结果元数据丢失:%v,错误=%v", columns, err)
}
if !rows.Next() {
t.Fatalf("没有读取到查询结果:%v", rows.Err())
}
var id, value int
if err := rows.Scan(&id, &value); err != nil || id != 1 || value != 0 {
t.Fatalf("查询数据错误:id=%d,value=%d,错误=%v", id, value, err)
}
if rows.Next() || rows.Err() != nil {
t.Fatalf("查询结束状态错误:%v", rows.Err())
}
}
func TestSQLiteOpenDeadlineWhileChangingJournalMode(t *testing.T) {
cfg := config.Default(t.TempDir()).Database
if err := ensureSQLiteDir(cfg.DSN); err != nil {
t.Fatal(err)
}
raw, err := sql.Open("sqlite", cfg.DSN)
if err != nil {
t.Fatal(err)
}
defer raw.Close()
if _, err := raw.ExecContext(t.Context(), "CREATE TABLE mode_probe (value INTEGER)"); err != nil {
t.Fatal(err)
}
writer, err := raw.BeginTx(t.Context(), nil)
if err != nil {
t.Fatal(err)
}
defer writer.Rollback()
if _, err := writer.ExecContext(t.Context(), "INSERT INTO mode_probe VALUES (7)"); err != nil {
t.Fatal(err)
}
ctx, cancel := context.WithTimeout(t.Context(), 150*time.Millisecond)
defer cancel()
started := time.Now()
db, err := Open(ctx, cfg)
if db != nil {
_ = db.Close()
}
if err == nil || time.Since(started) > time.Second {
t.Fatalf("初始化 WAL 遇到外部锁时没有及时失败:耗时=%s,错误=%v", time.Since(started), err)
}
if err := writer.Commit(); err != nil {
t.Fatal(err)
}
reopened := openSQLiteTestPool(t, cfg)
var value int
if err := reopened.QueryRowContext(t.Context(), "SELECT value FROM mode_probe").Scan(&value); err != nil || value != 7 {
t.Fatalf("初始化失败破坏了原有数据:%d,%v", value, err)
}
}

View File

@ -0,0 +1,56 @@
package database
import (
"context"
"errors"
"testing"
"time"
"modernc.org/sqlite"
)
func TestSQLiteNativeBusyWaitBoundaries(t *testing.T) {
for _, name := range []string{"busy_timeout", "manual_cancel_without_deadline"} {
t.Run(name, func(t *testing.T) {
first, second := newSQLiteConcurrencyPools(t)
writer, err := first.BeginTx(t.Context(), nil)
if err != nil {
t.Fatal(err)
}
defer writer.Rollback()
if _, err := writer.ExecContext(t.Context(), "UPDATE lock_probe SET value = 1 WHERE id = 1"); err != nil {
t.Fatal(err)
}
ctx := t.Context()
if name == "manual_cancel_without_deadline" {
var cancel context.CancelFunc
ctx, cancel = context.WithCancel(ctx)
defer cancel()
timer := time.AfterFunc(150*time.Millisecond, cancel)
defer timer.Stop()
}
started := time.Now()
_, err = second.ExecContext(ctx, "UPDATE lock_probe SET value = 99 WHERE id = 1")
elapsed := time.Since(started)
t.Logf("原生忙等待边界:%s,耗时=%s,错误=%v", name, elapsed, err)
if name == "busy_timeout" {
var sqliteErr *sqlite.Error
if !errors.As(err, &sqliteErr) || sqliteErr.Code() != 5 || elapsed < 4*time.Second {
t.Errorf("无截止时间的等待没有遵守默认忙等待:耗时=%s,错误=%v", elapsed, err)
}
} else if !errors.Is(err, context.Canceled) {
t.Errorf("手动取消后错误不正确:%v", err)
}
// 本测试记录现存原生等待上限,不代表无截止时间的取消可在一秒内完成。
if elapsed > 7*time.Second {
t.Errorf("原生忙等待超过允许的测试上限:%s", elapsed)
}
if err := writer.Rollback(); err != nil {
t.Fatal(err)
}
if value := readSQLiteProbe(t, second); value != 0 {
t.Fatalf("失败写入产生了副作用:%d", value)
}
})
}
}

View File

@ -0,0 +1,73 @@
package database
import (
"context"
"database/sql"
"database/sql/driver"
"fmt"
"modernc.org/sqlite"
)
func openSQLitePool(dsn string) *sql.DB {
return sql.OpenDB(&sqlitePoolConnector{driver: &sqlite.Driver{}, dsn: dsn})
}
type sqlitePoolConnector struct {
driver driver.Driver
dsn string
}
func (c *sqlitePoolConnector) Driver() driver.Driver {
return c.driver
}
func (c *sqlitePoolConnector) Connect(ctx context.Context) (driver.Conn, error) {
if err := ctx.Err(); err != nil {
return nil, err
}
conn, err := c.driver.Open(c.dsn)
if err != nil {
return nil, err
}
full, ok := conn.(sqliteDriverConn)
if !ok {
_ = conn.Close()
return nil, fmt.Errorf("SQLite 驱动连接接口不兼容")
}
wrapped := &sqliteDeadlineConn{sqliteDriverConn: full}
if err := configureSQLiteConnection(ctx, wrapped); err != nil {
_ = conn.Close()
return nil, err
}
return wrapped, nil
}
func configureSQLiteConnection(ctx context.Context, conn driver.Conn) error {
if err := ctx.Err(); err != nil {
return err
}
full, ok := conn.(sqlite.ExecQuerierContext)
if !ok {
return fmt.Errorf("SQLite 驱动连接接口不兼容")
}
// 必须在每条物理连接交给连接池之前设置,不能只配置池内恰好选中的连接。
if _, err := full.ExecContext(ctx, "PRAGMA busy_timeout = 5000", nil); err != nil {
return fmt.Errorf("配置 SQLite 忙等待失败:%w", err)
}
rows, err := full.QueryContext(ctx, "PRAGMA journal_mode = WAL", nil)
if err != nil {
return fmt.Errorf("配置 SQLite WAL 失败:%w", err)
}
defer rows.Close()
values := make([]driver.Value, 1)
if err := rows.Next(values); err != nil {
return fmt.Errorf("读取 SQLite 日志模式失败:%w", err)
}
if values[0] != "wal" && values[0] != "memory" {
return fmt.Errorf("SQLite 日志模式无效:%v", values[0])
}
return ctx.Err()
}
var _ driver.Connector = (*sqlitePoolConnector)(nil)

View File

@ -0,0 +1,77 @@
package database
import (
"context"
"database/sql"
"testing"
"git.pcux.cn/xuans/Scert/backend/internal/config"
)
func openSQLiteTestPool(t *testing.T, cfg config.DatabaseConfig) *sql.DB {
t.Helper()
db, err := Open(t.Context(), cfg)
if err != nil {
t.Fatalf("打开隔离 SQLite 文件库失败:%v", err)
}
t.Cleanup(func() {
if err := db.Close(); err != nil {
t.Errorf("关闭隔离 SQLite 连接池失败:%v", err)
}
})
return db.DB
}
func checkSQLiteConnection(t *testing.T, ctx context.Context, conn *sql.Conn) {
t.Helper()
var timeout int
if err := conn.QueryRowContext(ctx, "PRAGMA busy_timeout").Scan(&timeout); err != nil {
t.Fatalf("读取连接忙等待配置失败:%v", err)
}
if timeout != 5000 {
t.Errorf("连接忙等待配置丢失:得到 %d,期望 5000", timeout)
}
var mode string
if err := conn.QueryRowContext(ctx, "PRAGMA journal_mode").Scan(&mode); err != nil {
t.Fatalf("读取连接日志模式失败:%v", err)
}
if mode != "wal" {
t.Errorf("文件库没有使用 WAL:%s", mode)
}
}
func TestSQLitePoolConfiguresEveryConnection(t *testing.T) {
for _, name := range []string{"parallel_connections", "no_idle_connections", "replacement_connections", "uri_options"} {
t.Run(name, func(t *testing.T) {
cfg := config.Default(t.TempDir()).Database
cfg.MaxOpenConns = 4
cfg.MaxIdleConns = 4
if name == "no_idle_connections" {
cfg.MaxIdleConns = 0
}
if name == "uri_options" {
cfg.DSN += "?_pragma=foreign_keys(1)&_pragma=busy_timeout(17)&_txlock=immediate"
}
db := openSQLiteTestPool(t, cfg)
if name == "replacement_connections" {
db.SetMaxIdleConns(0)
db.SetMaxIdleConns(4)
}
// 同时持有所有连接,避免顺序查询反复使用初始化时的同一条连接。
for i := 0; i < 4; i++ {
conn, err := db.Conn(t.Context())
if err != nil {
t.Fatalf("取得第 %d 条连接失败:%v", i+1, err)
}
t.Cleanup(func() { _ = conn.Close() })
checkSQLiteConnection(t, t.Context(), conn)
if name == "uri_options" {
var enabled int
if err := conn.QueryRowContext(t.Context(), "PRAGMA foreign_keys").Scan(&enabled); err != nil || enabled != 1 {
t.Fatalf("用户提供的连接选项丢失:foreign_keys=%d,错误=%v", enabled, err)
}
}
}
})
}
}

View File

@ -0,0 +1,125 @@
package database
import (
"bufio"
"context"
"fmt"
"os"
"os/exec"
"strings"
"sync"
"testing"
"time"
"git.pcux.cn/xuans/Scert/backend/internal/config"
)
func TestSQLiteProcessConcurrency(t *testing.T) {
const writes = 50
if path := os.Getenv("SCERT_SQLITE_TEST_PROCESS_FILE"); path != "" {
cfg := config.Default(t.TempDir()).Database
cfg.DSN = path
db := openSQLiteTestPool(t, cfg)
fmt.Fprintln(os.Stdout, "sqlite-worker-ready")
for i := 0; i < writes; i++ {
if _, err := db.ExecContext(t.Context(), "UPDATE lock_probe SET value = value + 1 WHERE id = 1"); err != nil {
t.Fatalf("子进程第 %d 次写入失败:%v", i+1, err)
}
}
return
}
cfg := config.Default(t.TempDir()).Database
db := openSQLiteTestPool(t, cfg)
if _, err := db.ExecContext(t.Context(), "CREATE TABLE lock_probe (id INTEGER PRIMARY KEY, value INTEGER NOT NULL); INSERT INTO lock_probe VALUES (1, 0)"); err != nil {
t.Fatal(err)
}
var version string
if err := db.QueryRowContext(t.Context(), "SELECT sqlite_version()").Scan(&version); err != nil {
t.Fatal(err)
}
t.Logf("隔离多进程 SQLite 引擎版本:%s", version)
executable, err := os.Executable()
if err != nil {
t.Fatal(err)
}
ctx, cancel := context.WithTimeout(t.Context(), 30*time.Second)
defer cancel()
const processes = 3
results := make(chan error, processes)
ready := make(chan error, processes)
writer, err := db.BeginTx(t.Context(), nil)
if err != nil {
t.Fatal(err)
}
defer writer.Rollback()
if _, err := writer.ExecContext(t.Context(), "UPDATE lock_probe SET value = 0 WHERE id = 1"); err != nil {
t.Fatal(err)
}
var wg sync.WaitGroup
defer func() {
cancel()
wg.Wait()
}()
for i := 0; i < processes; i++ {
wg.Go(func() {
// 重用正式测试二进制,不生成脚本或启动项目外的数据库服务。
cmd := exec.CommandContext(ctx, executable, "-test.run=^TestSQLiteProcessConcurrency$", "-test.timeout=20s")
cmd.Env = append(os.Environ(), "SCERT_SQLITE_TEST_PROCESS_FILE="+cfg.DSN)
stdout, err := cmd.StdoutPipe()
if err != nil {
ready <- err
return
}
var stderr strings.Builder
cmd.Stderr = &stderr
if err := cmd.Start(); err != nil {
ready <- err
return
}
scanner := bufio.NewScanner(stdout)
signaled := false
for scanner.Scan() {
if scanner.Text() == "sqlite-worker-ready" && !signaled {
signaled = true
ready <- nil
}
}
waitErr := cmd.Wait()
if !signaled {
ready <- fmt.Errorf("子进程在就绪前退出:%v", waitErr)
}
if scanner.Err() != nil || waitErr != nil {
results <- fmt.Errorf("隔离子进程失败:等待=%v,读取=%v\n%s", waitErr, scanner.Err(), stderr.String())
}
})
}
// 三个独立进程均已打开数据库后再释放写锁,保证实际出现重叠的锁竞争。
for i := 0; i < processes; i++ {
select {
case err := <-ready:
if err != nil {
t.Error(err)
}
case <-ctx.Done():
t.Fatal("等待隔离子进程就绪超时")
}
}
if err := writer.Commit(); err != nil {
t.Fatal(err)
}
wg.Wait()
close(results)
for err := range results {
t.Error(err)
}
if value := readSQLiteProbe(t, db); value != processes*writes {
t.Fatalf("多进程提交总数错误:得到 %d,期望 %d", value, processes*writes)
}
if err := db.Close(); err != nil {
t.Fatal(err)
}
reopened := openSQLiteTestPool(t, cfg)
if value := readSQLiteProbe(t, reopened); value != processes*writes {
t.Fatalf("关闭连接后重新打开丢失数据:%d", value)
}
}