fix(数据库): 修复默认分组缺失与迁移锁阻塞

通过迁移补种默认 groups 记录,避免新装空分组
迁移锁改为 try lock + 重试并加入超时
写入 usage_logs 时保留 rate_multiplier=0 语义
测试: go test ./...
This commit is contained in:
yangjianbo
2025-12-29 10:43:46 +08:00
parent 3d617de577
commit 3a7d3387e0
5 changed files with 29 additions and 9 deletions

View File

@@ -10,6 +10,7 @@ import (
"io/fs"
"sort"
"strings"
"time"
"github.com/Wei-Shaw/sub2api/migrations"
)
@@ -31,6 +32,7 @@ CREATE TABLE IF NOT EXISTS schema_migrations (
// 在多实例部署场景下,该锁确保同一时间只有一个实例执行迁移。
// 任何稳定的 int64 值都可以,只要不与同一数据库中的其他锁冲突即可。
const migrationsAdvisoryLockID int64 = 694208311321144027
const migrationsLockRetryInterval = 500 * time.Millisecond
// ApplyMigrations 将嵌入的 SQL 迁移文件应用到指定的数据库。
//
@@ -166,11 +168,23 @@ func applyMigrationsFS(ctx context.Context, db *sql.DB, fsys fs.FS) error {
// Advisory Lock 是一种轻量级的锁机制,不与任何特定的数据库对象关联。
// 它非常适合用于应用层面的分布式锁场景,如迁移序列化。
func pgAdvisoryLock(ctx context.Context, db *sql.DB) error {
_, err := db.ExecContext(ctx, "SELECT pg_advisory_lock($1)", migrationsAdvisoryLockID)
if err != nil {
return fmt.Errorf("acquire migrations lock: %w", err)
ticker := time.NewTicker(migrationsLockRetryInterval)
defer ticker.Stop()
for {
var locked bool
if err := db.QueryRowContext(ctx, "SELECT pg_try_advisory_lock($1)", migrationsAdvisoryLockID).Scan(&locked); err != nil {
return fmt.Errorf("acquire migrations lock: %w", err)
}
if locked {
return nil
}
select {
case <-ctx.Done():
return fmt.Errorf("acquire migrations lock: %w", ctx.Err())
case <-ticker.C:
}
}
return nil
}
// pgAdvisoryUnlock 释放 PostgreSQL Advisory Lock。