126 lines
3.5 KiB
Go
126 lines
3.5 KiB
Go
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)
|
||
}
|
||
}
|