From af831aeea3199d788221f37d7411e6291d80ef8a Mon Sep 17 00:00:00 2001 From: xuans Date: Mon, 28 Sep 2026 01:06:01 +0800 Subject: [PATCH] =?UTF-8?q?=E4=BF=AE=E5=A4=8DSQLite=E8=BF=9E=E6=8E=A5?= =?UTF-8?q?=E9=85=8D=E7=BD=AE=E4=B8=8E=E6=88=AA=E6=AD=A2=E6=97=B6=E9=97=B4?= =?UTF-8?q?=E9=94=81=E7=AD=89=E5=BE=85=E5=B9=B6=E8=A1=A5=E5=85=85=E5=B9=B6?= =?UTF-8?q?=E5=8F=91=E5=9B=9E=E5=BD=92?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- .../cli/sqlite_snapshot_concurrency_test.go | 58 +++++ backend/internal/database/database.go | 26 +-- .../database/sqlite_concurrency_test.go | 208 ++++++++++++++++++ backend/internal/database/sqlite_deadline.go | 154 +++++++++++++ .../database/sqlite_deadline_fault_test.go | 79 +++++++ .../internal/database/sqlite_deadline_rows.go | 46 ++++ .../internal/database/sqlite_deadline_test.go | 186 ++++++++++++++++ .../database/sqlite_lock_boundary_test.go | 56 +++++ backend/internal/database/sqlite_pool.go | 73 ++++++ backend/internal/database/sqlite_pool_test.go | 77 +++++++ .../internal/database/sqlite_process_test.go | 125 +++++++++++ 11 files changed, 1064 insertions(+), 24 deletions(-) create mode 100644 backend/internal/cli/sqlite_snapshot_concurrency_test.go create mode 100644 backend/internal/database/sqlite_concurrency_test.go create mode 100644 backend/internal/database/sqlite_deadline.go create mode 100644 backend/internal/database/sqlite_deadline_fault_test.go create mode 100644 backend/internal/database/sqlite_deadline_rows.go create mode 100644 backend/internal/database/sqlite_deadline_test.go create mode 100644 backend/internal/database/sqlite_lock_boundary_test.go create mode 100644 backend/internal/database/sqlite_pool.go create mode 100644 backend/internal/database/sqlite_pool_test.go create mode 100644 backend/internal/database/sqlite_process_test.go diff --git a/backend/internal/cli/sqlite_snapshot_concurrency_test.go b/backend/internal/cli/sqlite_snapshot_concurrency_test.go new file mode 100644 index 0000000..c583fc8 --- /dev/null +++ b/backend/internal/cli/sqlite_snapshot_concurrency_test.go @@ -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) + } + }) + } +} diff --git a/backend/internal/database/database.go b/backend/internal/database/database.go index cfd543e..2397d49 100644 --- a/backend/internal/database/database.go +++ b/backend/internal/database/database.go @@ -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 { diff --git a/backend/internal/database/sqlite_concurrency_test.go b/backend/internal/database/sqlite_concurrency_test.go new file mode 100644 index 0000000..357e257 --- /dev/null +++ b/backend/internal/database/sqlite_concurrency_test.go @@ -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) + } +} diff --git a/backend/internal/database/sqlite_deadline.go b/backend/internal/database/sqlite_deadline.go new file mode 100644 index 0000000..053b47a --- /dev/null +++ b/backend/internal/database/sqlite_deadline.go @@ -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) diff --git a/backend/internal/database/sqlite_deadline_fault_test.go b/backend/internal/database/sqlite_deadline_fault_test.go new file mode 100644 index 0000000..2eacda7 --- /dev/null +++ b/backend/internal/database/sqlite_deadline_fault_test.go @@ -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) + } + } + }) + } +} diff --git a/backend/internal/database/sqlite_deadline_rows.go b/backend/internal/database/sqlite_deadline_rows.go new file mode 100644 index 0000000..2464160 --- /dev/null +++ b/backend/internal/database/sqlite_deadline_rows.go @@ -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) diff --git a/backend/internal/database/sqlite_deadline_test.go b/backend/internal/database/sqlite_deadline_test.go new file mode 100644 index 0000000..998a474 --- /dev/null +++ b/backend/internal/database/sqlite_deadline_test.go @@ -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) + } +} diff --git a/backend/internal/database/sqlite_lock_boundary_test.go b/backend/internal/database/sqlite_lock_boundary_test.go new file mode 100644 index 0000000..c249a3f --- /dev/null +++ b/backend/internal/database/sqlite_lock_boundary_test.go @@ -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) + } + }) + } +} diff --git a/backend/internal/database/sqlite_pool.go b/backend/internal/database/sqlite_pool.go new file mode 100644 index 0000000..b84ba52 --- /dev/null +++ b/backend/internal/database/sqlite_pool.go @@ -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) diff --git a/backend/internal/database/sqlite_pool_test.go b/backend/internal/database/sqlite_pool_test.go new file mode 100644 index 0000000..c7b0181 --- /dev/null +++ b/backend/internal/database/sqlite_pool_test.go @@ -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) + } + } + } + }) + } +} diff --git a/backend/internal/database/sqlite_process_test.go b/backend/internal/database/sqlite_process_test.go new file mode 100644 index 0000000..4cea049 --- /dev/null +++ b/backend/internal/database/sqlite_process_test.go @@ -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) + } +}