Merge remote-tracking branch 'origin/main' into feature/regulations-kb-ext20
This commit is contained in:
@@ -48,9 +48,11 @@ func main() {
|
||||
os.Exit(runCapabilities(os.Args[2:]))
|
||||
case "env":
|
||||
os.Exit(runEnv(os.Args[2:]))
|
||||
case "migrate":
|
||||
os.Exit(runMigrate(os.Args[2:]))
|
||||
default:
|
||||
fmt.Fprintf(os.Stderr, "agent-system: 未知子命令 %s\n", os.Args[1])
|
||||
fmt.Fprintln(os.Stderr, "用法:agent-system <serve|version|identity|user|apikey|config|capabilities|env> [参数]")
|
||||
fmt.Fprintln(os.Stderr, "用法:agent-system <serve|version|identity|user|apikey|config|capabilities|env|migrate> [参数]")
|
||||
os.Exit(2)
|
||||
}
|
||||
}
|
||||
@@ -77,6 +79,7 @@ func main() {
|
||||
pgIdleSec := fs.Int("pg-conn-max-idle-sec", 0, "PG 空闲连接回收秒数(0=默认 300)")
|
||||
pgLifetimeSec := fs.Int("pg-conn-max-lifetime-sec", 0, "PG 连接最长生命周期秒数(0=不限)")
|
||||
pgQueryTimeoutSec := fs.Int("pg-query-timeout-sec", 0, "PG statement_timeout 秒数(0=不限;覆盖全部业务查询)")
|
||||
adoptGuard := fs.Bool("adopt-instance-guard", false, "数据库被替换或锁文件被重建后,显式重写 SQLite 活动服务守卫")
|
||||
extension.RegisterCLIFlags(fs)
|
||||
fs.Parse(os.Args[1:])
|
||||
handled, hookErr := extension.RunCLIHooks(extension.Options{
|
||||
@@ -91,6 +94,7 @@ func main() {
|
||||
return
|
||||
}
|
||||
cliSet := visited(fs)
|
||||
delete(cliSet, "adopt-instance-guard")
|
||||
|
||||
resolved, err := config.Load(*cfgFile, os.LookupEnv, config.Config{
|
||||
Addr: *addr, DataDir: *data, PluginsDir: *pluginsDir, Mock: *mock,
|
||||
@@ -141,7 +145,8 @@ func main() {
|
||||
DisableConvCardMirror: extOpt.DisableConvCardMirror,
|
||||
DisableExecTools: extOpt.DisableExecTools, ExecBackend: extOpt.ExecBackend, ExecDepsDir: resolved.ExecDepsDir,
|
||||
ExecWorkspace: resolved.ExecWorkspace, ExecEnvRoot: resolved.ExecEnvRoot,
|
||||
Tier: resolved.Tier,
|
||||
Tier: resolved.Tier,
|
||||
AdoptInstanceGuard: *adoptGuard,
|
||||
PGPool: store.PoolOptions{
|
||||
MaxOpenConn: resolved.PGMaxOpen, MaxIdleConn: resolved.PGMaxIdle,
|
||||
ConnMaxIdleTimeSec: resolved.PGConnMaxIdleSec, ConnMaxLifetimeSec: resolved.PGConnMaxLifetimeSec,
|
||||
|
||||
@@ -0,0 +1,30 @@
|
||||
package main
|
||||
|
||||
import (
|
||||
"flag"
|
||||
"fmt"
|
||||
"os"
|
||||
|
||||
"agent-system-backend/internal/app"
|
||||
)
|
||||
|
||||
func runMigrate(args []string) int {
|
||||
fs := flag.NewFlagSet("migrate", flag.ExitOnError)
|
||||
data := fs.String("data", "./data", "数据目录(SQLite 档)")
|
||||
dsn := fs.String("dsn", "", "PostgreSQL DSN(空则读 AGENT_SYSTEM_DSN)")
|
||||
adopt := fs.Bool("adopt-instance-guard", false, "数据库被替换或锁文件被重建后,显式重写守卫")
|
||||
_ = fs.Parse(args)
|
||||
storage := *dsn
|
||||
if storage == "" {
|
||||
storage = os.Getenv("AGENT_SYSTEM_DSN")
|
||||
}
|
||||
if err := app.Migrate(app.Options{
|
||||
DataDir: *data,
|
||||
StorageDSN: storage,
|
||||
AdoptInstanceGuard: *adopt,
|
||||
}); err != nil {
|
||||
fmt.Fprintln(os.Stderr, "agent-system:", err)
|
||||
return 1
|
||||
}
|
||||
return 0
|
||||
}
|
||||
@@ -300,6 +300,8 @@ var copyOrder = []string{
|
||||
"recovery_proposals", "recovery_proposal_proofs",
|
||||
// v35 分层 Python 环境(无外键;申请与会话绑定要随库迁走)
|
||||
"exec_env_requests", "exec_env_bindings",
|
||||
// U8:活动服务锁的库内标记(无外键;迁移须保留以免目标库丢掉锁根)
|
||||
"instance_guard",
|
||||
}
|
||||
|
||||
const controlDDL = `
|
||||
|
||||
Vendored
+1
-1
Submodule deps/development-manager updated: bd3a7de57a...67784f43eb
+5
-42
@@ -30,7 +30,6 @@ import (
|
||||
"agent-native/runtime/telemetry"
|
||||
agenttools "agent-native/runtime/tools"
|
||||
nativeadp "agent-squad/adapters/native"
|
||||
"agent-system-backend/internal/blobstore"
|
||||
"agent-system-backend/internal/bus"
|
||||
"agent-system-backend/internal/cardsvc"
|
||||
"agent-system-backend/internal/cmdbus"
|
||||
@@ -97,6 +96,8 @@ type Options struct {
|
||||
Tier string
|
||||
// PGPool PG 连接预算(KTD12;零值走默认)。仅 PG 档生效。
|
||||
PGPool store.PoolOptions
|
||||
// AdoptInstanceGuard 在数据库被替换或锁文件被重建后,显式重写守卫记录。
|
||||
AdoptInstanceGuard bool
|
||||
}
|
||||
|
||||
// instanceTag 稳定执行身份:tier=team 时强制 host:addr(多实例互异),
|
||||
@@ -175,49 +176,11 @@ func Run(opt Options) error {
|
||||
if opt.InsecureListen && !config.IsLoopbackListen(opt.Addr) && !resolved.AuthEnabled() {
|
||||
log.Printf("warning: listening on %s without authentication because insecureListen is set", opt.Addr)
|
||||
}
|
||||
var st *store.Store
|
||||
var err error
|
||||
if opt.StorageDSN != "" {
|
||||
artifacts := opt.ArtifactsDir
|
||||
if artifacts == "" {
|
||||
artifacts = filepath.Join(opt.DataDir, "artifacts")
|
||||
}
|
||||
// 共享工件(U5/KTD9):配置了 S3 兼容对象存储时注入(多实例共享);
|
||||
// 未配置回落本地目录(单实例/共享卷)。凭据只经环境(不入库不入报告)。
|
||||
blobs, berr := blobstore.NewS3FromEnv()
|
||||
if berr != nil {
|
||||
return fmt.Errorf("blobstore s3: %w", berr)
|
||||
}
|
||||
if blobs != nil {
|
||||
log.Printf("artifacts: shared via S3-compatible object storage (prefix %q)", blobs.Prefix)
|
||||
}
|
||||
st, err = store.OpenPostgresWithOptions(opt.StorageDSN, artifacts, opt.PGPool, blobs)
|
||||
if err != nil && opt.Tier == "team" {
|
||||
// team 档拒绝就绪(R2):PG 故障不回落 SQLite,错误必须明说。
|
||||
return fmt.Errorf("team 档打开 PostgreSQL 失败,拒绝就绪(不回落 SQLite): %w", err)
|
||||
}
|
||||
} else {
|
||||
st, err = store.OpenSQLite(filepath.Join(opt.DataDir, "agent-system.db"))
|
||||
}
|
||||
st, releaseStore, err := openServiceStore(opt)
|
||||
if err != nil {
|
||||
return fmt.Errorf("open store: %w", err)
|
||||
}
|
||||
defer st.Close()
|
||||
// U1 I3:svc 注入的服务实例身份(manifest.instance_id)与库内 marker 对齐;
|
||||
// 未设置则跳过(兼容未接线旧调用方)。与 --instance(runs 进程标识)无关。
|
||||
if sysID := strings.TrimSpace(os.Getenv("AGENT_SYSTEM_INSTANCE_ID")); sysID != "" {
|
||||
dbref := strings.TrimSpace(os.Getenv("AGENT_SYSTEM_DBREF"))
|
||||
if dbref == "" {
|
||||
if opt.StorageDSN != "" {
|
||||
dbref = "pg"
|
||||
} else {
|
||||
dbref = "sqlite"
|
||||
}
|
||||
}
|
||||
if err := st.VerifySystemIdentity(sysID, dbref); err != nil {
|
||||
return fmt.Errorf("system identity: %w", err)
|
||||
}
|
||||
return err
|
||||
}
|
||||
defer releaseStore()
|
||||
pool := st.Pool()
|
||||
if versions, verr := st.MigrationStatus(); verr == nil {
|
||||
log.Printf("storage tier: %s (dialect=%s pool=%d/%d idleSec=%d lifetimeSec=%d schema=v%d)",
|
||||
|
||||
@@ -0,0 +1,201 @@
|
||||
//go:build unix
|
||||
|
||||
package app
|
||||
|
||||
import (
|
||||
"errors"
|
||||
"os"
|
||||
"os/exec"
|
||||
"path/filepath"
|
||||
"testing"
|
||||
"time"
|
||||
|
||||
"agent-system-backend/internal/instanceguard"
|
||||
"agent-system-backend/internal/store"
|
||||
core "agent-system-core"
|
||||
)
|
||||
|
||||
func TestSecondProcessRejectedBeforeOpen(t *testing.T) {
|
||||
if os.Getenv("U8_APP_ROLE") == "hold" {
|
||||
holdService(t)
|
||||
return
|
||||
}
|
||||
dir := t.TempDir()
|
||||
ready := filepath.Join(dir, "ready")
|
||||
cmd := exec.Command(os.Args[0], "-test.run=TestSecondProcessRejectedBeforeOpen")
|
||||
cmd.Env = append(os.Environ(), "U8_APP_ROLE=hold", "U8_APP_DIR="+dir, "U8_APP_READY="+ready)
|
||||
cmd.Dir = t.TempDir()
|
||||
cmd.Stdout = os.Stdout
|
||||
cmd.Stderr = os.Stderr
|
||||
if err := cmd.Start(); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
t.Cleanup(func() {
|
||||
if cmd.Process != nil {
|
||||
_ = cmd.Process.Kill()
|
||||
_, _ = cmd.Process.Wait()
|
||||
}
|
||||
})
|
||||
waitReady(t, ready)
|
||||
_, _, err := openServiceStore(Options{DataDir: dir, Addr: ":1"})
|
||||
if !errors.Is(err, instanceguard.ErrBusy) {
|
||||
t.Fatalf("second open: %v", err)
|
||||
}
|
||||
if err := Migrate(Options{DataDir: dir}); !errors.Is(err, instanceguard.ErrBusy) {
|
||||
t.Fatalf("migrate while held: %v", err)
|
||||
}
|
||||
cli, err := store.OpenExistingSQLite(filepath.Join(dir, "agent-system.db"))
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
run, err := cli.GetRunByID("run_u8_hold")
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if run.Status != core.RunWaitingHITL {
|
||||
t.Fatalf("run changed while second process was refused: %s", run.Status)
|
||||
}
|
||||
if _, err := cli.CreateProject("owner", "short"); err != nil {
|
||||
t.Fatalf("account-style write while service holds the guard: %v", err)
|
||||
}
|
||||
if err := cli.Close(); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if err := cmd.Process.Kill(); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
_, _ = cmd.Process.Wait()
|
||||
st, release, err := openServiceStore(Options{DataDir: dir, Addr: ":2"})
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
defer release()
|
||||
run, err = st.GetRunByID("run_u8_hold")
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if run.Status != core.RunWaitingHITL {
|
||||
t.Fatalf("restart reconciled a run the second process never opened: %s", run.Status)
|
||||
}
|
||||
}
|
||||
|
||||
func TestCopiedBinaryStillSharesGuard(t *testing.T) {
|
||||
if os.Getenv("U8_APP_ROLE") == "copy-hold" {
|
||||
holdService(t)
|
||||
return
|
||||
}
|
||||
bin, err := os.Executable()
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
copyPath := filepath.Join(t.TempDir(), "agent-system-test-copy")
|
||||
raw, err := os.ReadFile(bin)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if err := os.WriteFile(copyPath, raw, 0o700); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
dir := t.TempDir()
|
||||
ready := filepath.Join(dir, "ready")
|
||||
cmd := exec.Command(copyPath, "-test.run=TestCopiedBinaryStillSharesGuard")
|
||||
cmd.Env = append(os.Environ(), "U8_APP_ROLE=copy-hold", "U8_APP_DIR="+dir, "U8_APP_READY="+ready)
|
||||
cmd.Dir = t.TempDir()
|
||||
cmd.Stdout = os.Stdout
|
||||
cmd.Stderr = os.Stderr
|
||||
if err := cmd.Start(); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
t.Cleanup(func() {
|
||||
if cmd.Process != nil {
|
||||
_ = cmd.Process.Kill()
|
||||
_, _ = cmd.Process.Wait()
|
||||
}
|
||||
})
|
||||
waitReady(t, ready)
|
||||
if _, _, err := openServiceStore(Options{DataDir: dir}); !errors.Is(err, instanceguard.ErrBusy) {
|
||||
t.Fatalf("copied binary: %v", err)
|
||||
}
|
||||
}
|
||||
|
||||
func holdService(t *testing.T) {
|
||||
t.Helper()
|
||||
dir := os.Getenv("U8_APP_DIR")
|
||||
st, _, err := openServiceStore(Options{DataDir: dir})
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
run := &core.Run{
|
||||
ID: "run_u8_hold", ConversationID: "c", ProjectID: "p", Owner: "o",
|
||||
Kind: core.RunConversation, Status: core.RunWaitingHITL,
|
||||
}
|
||||
if err := st.CreateRun(run); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if err := os.WriteFile(os.Getenv("U8_APP_READY"), []byte("ready"), 0o600); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
time.Sleep(30 * time.Second)
|
||||
}
|
||||
|
||||
func waitReady(t *testing.T, ready string) {
|
||||
t.Helper()
|
||||
deadline := time.Now().Add(20 * time.Second)
|
||||
for time.Now().Before(deadline) {
|
||||
if _, err := os.Stat(ready); err == nil {
|
||||
return
|
||||
}
|
||||
time.Sleep(20 * time.Millisecond)
|
||||
}
|
||||
t.Fatal("child did not become ready")
|
||||
}
|
||||
|
||||
func TestAdoptDoesNotCreateMissingDatabase(t *testing.T) {
|
||||
dir := t.TempDir()
|
||||
_, release, err := openServiceStore(Options{DataDir: dir})
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
release()
|
||||
db := filepath.Join(dir, "agent-system.db")
|
||||
if err := os.Remove(db); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
_, _, err = openServiceStore(Options{DataDir: dir, AdoptInstanceGuard: true})
|
||||
if !errors.Is(err, instanceguard.ErrDBMissing) {
|
||||
t.Fatalf("adopt missing: %v", err)
|
||||
}
|
||||
if _, err := os.Stat(db); !os.IsNotExist(err) {
|
||||
t.Fatalf("adopt created a database: %v", err)
|
||||
}
|
||||
}
|
||||
|
||||
func TestIdentityFailureDoesNotStick(t *testing.T) {
|
||||
dir := t.TempDir()
|
||||
t.Setenv("AGENT_SYSTEM_INSTANCE_ID", "wrong")
|
||||
if _, _, err := openServiceStore(Options{DataDir: dir}); err == nil {
|
||||
t.Fatal("missing identity row was accepted")
|
||||
}
|
||||
t.Setenv("AGENT_SYSTEM_INSTANCE_ID", "")
|
||||
st, err := store.OpenSQLite(filepath.Join(dir, "agent-system.db"))
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
marker, ok, err := st.LoadInstanceGuard()
|
||||
if err != nil || ok || marker.InstanceID != "" {
|
||||
t.Fatalf("failed start wrote a guard id: ok=%v id=%q err=%v", ok, marker.InstanceID, err)
|
||||
}
|
||||
if err := st.EnsureSystemIdentity("right", "sqlite"); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if err := st.Close(); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
t.Setenv("AGENT_SYSTEM_INSTANCE_ID", "right")
|
||||
opened, release, err := openServiceStore(Options{DataDir: dir})
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
release()
|
||||
_ = opened
|
||||
}
|
||||
@@ -0,0 +1,111 @@
|
||||
package app
|
||||
|
||||
import (
|
||||
"errors"
|
||||
"fmt"
|
||||
"log"
|
||||
"os"
|
||||
"path/filepath"
|
||||
"strings"
|
||||
|
||||
"agent-system-backend/internal/blobstore"
|
||||
"agent-system-backend/internal/instanceguard"
|
||||
"agent-system-backend/internal/store"
|
||||
)
|
||||
|
||||
// openServiceStore 打开活动服务使用的存储。
|
||||
// SQLite 档先取活动服务锁,再打开(打开即迁移),然后核对守卫标记。
|
||||
// 返回的函数先关存储再放锁。
|
||||
func openServiceStore(opt Options) (*store.Store, func(), error) {
|
||||
nop := func() {}
|
||||
if opt.StorageDSN != "" {
|
||||
artifacts := opt.ArtifactsDir
|
||||
if artifacts == "" {
|
||||
artifacts = filepath.Join(opt.DataDir, "artifacts")
|
||||
}
|
||||
// 共享工件(U5/KTD9):配置了 S3 兼容对象存储时注入(多实例共享);
|
||||
// 未配置回落本地目录(单实例/共享卷)。凭据只经环境(不入库不入报告)。
|
||||
blobs, berr := blobstore.NewS3FromEnv()
|
||||
if berr != nil {
|
||||
return nil, nop, fmt.Errorf("blobstore s3: %w", berr)
|
||||
}
|
||||
if blobs != nil {
|
||||
log.Printf("artifacts: shared via S3-compatible object storage (prefix %q)", blobs.Prefix)
|
||||
}
|
||||
st, err := store.OpenPostgresWithOptions(opt.StorageDSN, artifacts, opt.PGPool, blobs)
|
||||
if err != nil && opt.Tier == "team" {
|
||||
// team 档拒绝就绪(R2):PG 故障不回落 SQLite,错误必须明说。
|
||||
return nil, nop, fmt.Errorf("team 档打开 PostgreSQL 失败,拒绝就绪(不回落 SQLite): %w", err)
|
||||
}
|
||||
if err != nil {
|
||||
return nil, nop, fmt.Errorf("open store: %w", err)
|
||||
}
|
||||
if err := verifyConfiguredIdentity(st, "pg"); err != nil {
|
||||
_ = st.Close()
|
||||
return nil, nop, err
|
||||
}
|
||||
return st, func() { _ = st.Close() }, nil
|
||||
}
|
||||
|
||||
lease, err := instanceguard.Acquire(opt.DataDir)
|
||||
if err != nil {
|
||||
return nil, nop, fmt.Errorf("instance guard: %w", err)
|
||||
}
|
||||
releaseLease := func() { lease.Release() }
|
||||
if err := lease.AutomaticRefusal(); err != nil {
|
||||
adoptable := opt.AdoptInstanceGuard && (errors.Is(err, instanceguard.ErrDBReplaced) || errors.Is(err, instanceguard.ErrLockCorrupt))
|
||||
if !adoptable {
|
||||
releaseLease()
|
||||
return nil, nop, fmt.Errorf("instance guard: %w", err)
|
||||
}
|
||||
}
|
||||
st, err := store.OpenSQLite(lease.DBPath())
|
||||
if err != nil {
|
||||
releaseLease()
|
||||
return nil, nop, fmt.Errorf("open store: %w", err)
|
||||
}
|
||||
if err := verifyConfiguredIdentity(st, "sqlite"); err != nil {
|
||||
_ = st.Close()
|
||||
releaseLease()
|
||||
return nil, nop, err
|
||||
}
|
||||
if err := instanceguard.Bind(st, lease, instanceguard.BindOptions{
|
||||
Adopt: opt.AdoptInstanceGuard,
|
||||
InstanceID: strings.TrimSpace(os.Getenv("AGENT_SYSTEM_INSTANCE_ID")),
|
||||
}); err != nil {
|
||||
_ = st.Close()
|
||||
releaseLease()
|
||||
return nil, nop, fmt.Errorf("instance guard: %w", err)
|
||||
}
|
||||
return st, func() {
|
||||
_ = st.Close()
|
||||
lease.Release()
|
||||
}, nil
|
||||
}
|
||||
|
||||
// verifyConfiguredIdentity U1 I3:svc 注入的服务实例身份(manifest.instance_id)与库内 marker 对齐;
|
||||
// 未设置则跳过(兼容未接线旧调用方)。与 --instance(runs 进程标识)无关。
|
||||
func verifyConfiguredIdentity(st *store.Store, defaultRef string) error {
|
||||
sysID := strings.TrimSpace(os.Getenv("AGENT_SYSTEM_INSTANCE_ID"))
|
||||
if sysID == "" {
|
||||
return nil
|
||||
}
|
||||
dbref := strings.TrimSpace(os.Getenv("AGENT_SYSTEM_DBREF"))
|
||||
if dbref == "" {
|
||||
dbref = defaultRef
|
||||
}
|
||||
if err := st.VerifySystemIdentity(sysID, dbref); err != nil {
|
||||
return fmt.Errorf("system identity: %w", err)
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
// Migrate 执行显式迁移。SQLite 必须先取得活动服务锁,有服务占用则拒绝。
|
||||
func Migrate(opt Options) error {
|
||||
_, release, err := openServiceStore(opt)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
release()
|
||||
return nil
|
||||
}
|
||||
@@ -0,0 +1,199 @@
|
||||
// Package instanceguard 强制同一 SQLite 数据文件只有一个活动服务。
|
||||
// 锁在打开数据库(迁移发生在打开内部)之前取得。账户 CLI 不使用本包。
|
||||
package instanceguard
|
||||
|
||||
import (
|
||||
"crypto/rand"
|
||||
"encoding/hex"
|
||||
"errors"
|
||||
"strconv"
|
||||
"strings"
|
||||
|
||||
"agent-system-backend/internal/store"
|
||||
)
|
||||
|
||||
var (
|
||||
ErrBusy = errors.New("instance guard: sqlite service already running")
|
||||
ErrHardlink = errors.New("instance guard: sqlite database has a hardlink alias")
|
||||
ErrDBReplaced = errors.New("instance guard: sqlite database replaced")
|
||||
ErrDBMissing = errors.New("instance guard: recorded sqlite database is missing")
|
||||
ErrLockRebuilt = errors.New("instance guard: lock file was rebuilt")
|
||||
ErrLockCorrupt = errors.New("instance guard: lock record is corrupt")
|
||||
ErrInstanceMismatch = errors.New("instance guard: instance id mismatch")
|
||||
ErrUnsupported = errors.New("instance guard: sqlite activity lock is not available on this platform")
|
||||
ErrLockAliased = errors.New("instance guard: lock file aliases the sqlite database")
|
||||
)
|
||||
|
||||
// Record 是锁文件里记下的数据库身份。没有记录表示锁文件是新建的。
|
||||
type Record struct {
|
||||
DBDev uint64
|
||||
DBIno uint64
|
||||
InstanceID string
|
||||
LockGen string
|
||||
}
|
||||
|
||||
// Lease 持有活动服务锁,直到 Release。关闭顺序由调用方保证:先关存储,再 Release。
|
||||
type Lease struct {
|
||||
fd int
|
||||
dbPath string
|
||||
lockDev uint64
|
||||
lockIno uint64
|
||||
dbDev uint64
|
||||
dbIno uint64
|
||||
dbExists bool
|
||||
record *Record
|
||||
parseErr error
|
||||
release func()
|
||||
}
|
||||
|
||||
// Release 放开锁并关闭锁文件描述符。锁文件留在磁盘上。
|
||||
func (l *Lease) Release() {
|
||||
if l == nil || l.release == nil {
|
||||
return
|
||||
}
|
||||
l.release()
|
||||
l.release = nil
|
||||
}
|
||||
|
||||
// DBPath 是解析后的数据库路径。符号链接指向真实文件。
|
||||
func (l *Lease) DBPath() string {
|
||||
if l == nil {
|
||||
return ""
|
||||
}
|
||||
return l.dbPath
|
||||
}
|
||||
|
||||
// AutomaticRefusal 在打开数据库之前调用。
|
||||
// 尚无记录时返回 nil(首次建立,含旧库升级)。硬链接在 Acquire 阶段已经拒绝。
|
||||
func (l *Lease) AutomaticRefusal() error {
|
||||
if l.parseErr != nil {
|
||||
return l.parseErr
|
||||
}
|
||||
if l.record == nil {
|
||||
return nil
|
||||
}
|
||||
if !l.dbExists {
|
||||
return ErrDBMissing
|
||||
}
|
||||
if l.record.DBDev != l.dbDev || l.record.DBIno != l.dbIno {
|
||||
return ErrDBReplaced
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
// BindOptions 控制守卫建立后的比对。
|
||||
type BindOptions struct {
|
||||
Adopt bool
|
||||
InstanceID string
|
||||
}
|
||||
|
||||
// Bind 在数据库已经打开并完成迁移之后调用。
|
||||
// 锁文件 inode 与表不一致则拒绝,除非 Adopt。通过时重写锁文件内容与表。
|
||||
func Bind(st *store.Store, lease *Lease, opt BindOptions) error {
|
||||
if lease == nil {
|
||||
return errors.New("instance guard: nil lease")
|
||||
}
|
||||
marker, ok, err := st.LoadInstanceGuard()
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
if ok && !opt.Adopt {
|
||||
inodeMismatch := marker.LockDev != lease.lockDev || marker.LockIno != lease.lockIno
|
||||
genMismatch := marker.LockGen != "" && (lease.record == nil || lease.record.LockGen != marker.LockGen)
|
||||
if inodeMismatch || genMismatch {
|
||||
return ErrLockRebuilt
|
||||
}
|
||||
}
|
||||
id := opt.InstanceID
|
||||
if ok && marker.InstanceID != "" && id != "" && marker.InstanceID != id && !opt.Adopt {
|
||||
return ErrInstanceMismatch
|
||||
}
|
||||
if id == "" && ok {
|
||||
id = marker.InstanceID
|
||||
}
|
||||
dev, ino, exists, err := lease.currentDB()
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
if !exists {
|
||||
return ErrDBMissing
|
||||
}
|
||||
lease.dbDev, lease.dbIno, lease.dbExists = dev, ino, true
|
||||
gen := ""
|
||||
if ok && marker.LockGen != "" && !opt.Adopt {
|
||||
gen = marker.LockGen
|
||||
} else {
|
||||
gen, err = newLockGen()
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
}
|
||||
next := store.InstanceGuardMarker{
|
||||
DBDev: dev, DBIno: ino, LockDev: lease.lockDev, LockIno: lease.lockIno,
|
||||
InstanceID: id, LockGen: gen,
|
||||
}
|
||||
if err := lease.writeRecord(dev, ino, id, gen); err != nil {
|
||||
return err
|
||||
}
|
||||
if !ok || opt.Adopt || marker != next {
|
||||
if err := st.SaveInstanceGuard(next); err != nil {
|
||||
return err
|
||||
}
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
func newLockGen() (string, error) {
|
||||
buf := make([]byte, 16)
|
||||
if _, err := rand.Read(buf); err != nil {
|
||||
return "", err
|
||||
}
|
||||
return hex.EncodeToString(buf), nil
|
||||
}
|
||||
|
||||
func parseRecord(raw string) (*Record, error) {
|
||||
raw = strings.TrimSpace(raw)
|
||||
if raw == "" {
|
||||
return nil, nil
|
||||
}
|
||||
parts := strings.Split(raw, "\t")
|
||||
if parts[0] != "v1" {
|
||||
return nil, ErrLockCorrupt
|
||||
}
|
||||
rec := &Record{}
|
||||
seen := map[string]bool{}
|
||||
for _, part := range parts[1:] {
|
||||
key, val, ok := strings.Cut(part, "=")
|
||||
if !ok || seen[key] || strings.ContainsAny(val, " \t") {
|
||||
return nil, ErrLockCorrupt
|
||||
}
|
||||
seen[key] = true
|
||||
switch key {
|
||||
case "db_dev":
|
||||
n, err := strconv.ParseUint(val, 10, 64)
|
||||
if err != nil {
|
||||
return nil, ErrLockCorrupt
|
||||
}
|
||||
rec.DBDev = n
|
||||
case "db_ino":
|
||||
n, err := strconv.ParseUint(val, 10, 64)
|
||||
if err != nil {
|
||||
return nil, ErrLockCorrupt
|
||||
}
|
||||
rec.DBIno = n
|
||||
case "instance_id":
|
||||
rec.InstanceID = val
|
||||
case "lock_gen":
|
||||
if val == "" {
|
||||
return nil, ErrLockCorrupt
|
||||
}
|
||||
rec.LockGen = val
|
||||
default:
|
||||
return nil, ErrLockCorrupt
|
||||
}
|
||||
}
|
||||
if !seen["db_dev"] || !seen["db_ino"] || !seen["instance_id"] {
|
||||
return nil, ErrLockCorrupt
|
||||
}
|
||||
return rec, nil
|
||||
}
|
||||
@@ -0,0 +1,18 @@
|
||||
//go:build !unix
|
||||
|
||||
package instanceguard
|
||||
|
||||
import "fmt"
|
||||
|
||||
// Acquire 在非 unix 平台拒绝 SQLite 活动服务。PostgreSQL 档不调用本函数。
|
||||
func Acquire(string) (*Lease, error) {
|
||||
return nil, ErrUnsupported
|
||||
}
|
||||
|
||||
func (l *Lease) writeRecord(uint64, uint64, string, string) error {
|
||||
return ErrUnsupported
|
||||
}
|
||||
|
||||
func (l *Lease) currentDB() (uint64, uint64, bool, error) {
|
||||
return 0, 0, false, fmt.Errorf("%w", ErrUnsupported)
|
||||
}
|
||||
@@ -0,0 +1,406 @@
|
||||
//go:build unix
|
||||
|
||||
package instanceguard
|
||||
|
||||
import (
|
||||
"errors"
|
||||
"os"
|
||||
"os/exec"
|
||||
"path/filepath"
|
||||
"testing"
|
||||
"time"
|
||||
|
||||
"agent-system-backend/internal/store"
|
||||
)
|
||||
|
||||
func TestAcquireSameResourceAcrossAliases(t *testing.T) {
|
||||
base := t.TempDir()
|
||||
real := filepath.Join(base, "real")
|
||||
if err := os.Mkdir(real, 0o755); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
link := filepath.Join(base, "link")
|
||||
if err := os.Symlink(real, link); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
held, err := Acquire(link)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
defer held.Release()
|
||||
if _, err := Acquire(real); !errors.Is(err, ErrBusy) {
|
||||
t.Fatalf("symlink alias: %v", err)
|
||||
}
|
||||
|
||||
wd, err := os.Getwd()
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if err := os.Chdir(base); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
t.Cleanup(func() { _ = os.Chdir(wd) })
|
||||
if _, err := Acquire(filepath.Join(".", "real")); !errors.Is(err, ErrBusy) {
|
||||
t.Fatalf("relative path: %v", err)
|
||||
}
|
||||
}
|
||||
|
||||
func TestHardlinkRejected(t *testing.T) {
|
||||
dir := t.TempDir()
|
||||
db := filepath.Join(dir, "agent-system.db")
|
||||
st, err := store.OpenSQLite(db)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if err := st.Close(); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if err := os.Link(db, filepath.Join(dir, "alias.db")); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if _, err := Acquire(dir); !errors.Is(err, ErrHardlink) {
|
||||
t.Fatalf("hardlink: %v", err)
|
||||
}
|
||||
}
|
||||
|
||||
func TestDatabaseReplacementRefusedBeforeOpen(t *testing.T) {
|
||||
dir := t.TempDir()
|
||||
st, release, err := openAndBind(t, dir, "")
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
db := filepath.Join(dir, "agent-system.db")
|
||||
release()
|
||||
_ = st
|
||||
if err := os.Remove(db); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
raw, err := os.ReadFile(db + ".lock")
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
rec, err := parseRecord(string(raw))
|
||||
if err != nil || rec == nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
for i := 0; i < 16; i++ {
|
||||
if err := os.WriteFile(db, []byte("replaced"), 0o600); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
_, ino, exists, err := statFile(db)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if exists && ino != rec.DBIno {
|
||||
break
|
||||
}
|
||||
if err := os.Remove(db); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if err := os.WriteFile(filepath.Join(dir, "inode-burn-"+string(rune('a'+i))), []byte("x"), 0o600); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
}
|
||||
lease, err := Acquire(dir)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
defer lease.Release()
|
||||
if err := lease.AutomaticRefusal(); !errors.Is(err, ErrDBReplaced) {
|
||||
t.Fatalf("replaced db: %v", err)
|
||||
}
|
||||
if _, err := os.Stat(db); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
}
|
||||
|
||||
func TestMissingRecordedDatabaseRefused(t *testing.T) {
|
||||
dir := t.TempDir()
|
||||
_, release, err := openAndBind(t, dir, "")
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
release()
|
||||
db := filepath.Join(dir, "agent-system.db")
|
||||
if err := os.Remove(db); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
lease, err := Acquire(dir)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
defer lease.Release()
|
||||
if err := lease.AutomaticRefusal(); !errors.Is(err, ErrDBMissing) {
|
||||
t.Fatalf("missing db: %v", err)
|
||||
}
|
||||
}
|
||||
|
||||
func TestRebuiltLockRefusedUntilAdopt(t *testing.T) {
|
||||
dir := t.TempDir()
|
||||
st, release, err := openAndBind(t, dir, "inst-a")
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
release()
|
||||
_ = st.Close()
|
||||
if err := os.Remove(filepath.Join(dir, "agent-system.db.lock")); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
lease, err := Acquire(dir)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if err := lease.AutomaticRefusal(); err != nil {
|
||||
lease.Release()
|
||||
t.Fatal(err)
|
||||
}
|
||||
opened, err := store.OpenSQLite(lease.DBPath())
|
||||
if err != nil {
|
||||
lease.Release()
|
||||
t.Fatal(err)
|
||||
}
|
||||
err = Bind(opened, lease, BindOptions{InstanceID: "inst-a"})
|
||||
_ = opened.Close()
|
||||
lease.Release()
|
||||
if !errors.Is(err, ErrLockRebuilt) {
|
||||
t.Fatalf("rebuilt lock: %v", err)
|
||||
}
|
||||
|
||||
adopted, err := Acquire(dir)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
defer adopted.Release()
|
||||
opened, err = store.OpenSQLite(adopted.DBPath())
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
defer opened.Close()
|
||||
if err := Bind(opened, adopted, BindOptions{Adopt: true, InstanceID: "inst-a"}); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
}
|
||||
|
||||
func TestClearedLockRecordRefusedWhenInodeStays(t *testing.T) {
|
||||
dir := t.TempDir()
|
||||
st, release, err := openAndBind(t, dir, "inst-a")
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
release()
|
||||
_ = st.Close()
|
||||
lockPath := filepath.Join(dir, "agent-system.db.lock")
|
||||
if err := os.Truncate(lockPath, 0); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
lease, err := Acquire(dir)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
opened, err := store.OpenSQLite(lease.DBPath())
|
||||
if err != nil {
|
||||
lease.Release()
|
||||
t.Fatal(err)
|
||||
}
|
||||
err = Bind(opened, lease, BindOptions{InstanceID: "inst-a"})
|
||||
_ = opened.Close()
|
||||
lease.Release()
|
||||
if !errors.Is(err, ErrLockRebuilt) {
|
||||
t.Fatalf("cleared lock: %v", err)
|
||||
}
|
||||
}
|
||||
|
||||
func TestInstanceMismatchPreservesEmpty(t *testing.T) {
|
||||
dir := t.TempDir()
|
||||
st, release, err := openAndBind(t, dir, "inst-a")
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
release()
|
||||
_ = st.Close()
|
||||
|
||||
lease, err := Acquire(dir)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
opened, err := store.OpenSQLite(lease.DBPath())
|
||||
if err != nil {
|
||||
lease.Release()
|
||||
t.Fatal(err)
|
||||
}
|
||||
err = Bind(opened, lease, BindOptions{InstanceID: "inst-b"})
|
||||
_ = opened.Close()
|
||||
lease.Release()
|
||||
if !errors.Is(err, ErrInstanceMismatch) {
|
||||
t.Fatalf("mismatch: %v", err)
|
||||
}
|
||||
|
||||
lease, err = Acquire(dir)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
defer lease.Release()
|
||||
opened, err = store.OpenSQLite(lease.DBPath())
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
defer opened.Close()
|
||||
if err := Bind(opened, lease, BindOptions{}); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
marker, ok, err := opened.LoadInstanceGuard()
|
||||
if err != nil || !ok || marker.InstanceID != "inst-a" {
|
||||
t.Fatalf("empty id should keep marker: ok=%v id=%q err=%v", ok, marker.InstanceID, err)
|
||||
}
|
||||
}
|
||||
|
||||
func TestLockReleasedAfterKill(t *testing.T) {
|
||||
if os.Getenv("U8_GUARD_ROLE") == "hold" {
|
||||
lease, err := Acquire(os.Getenv("U8_GUARD_DIR"))
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if err := os.WriteFile(os.Getenv("U8_GUARD_READY"), []byte("ready"), 0o600); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
time.Sleep(30 * time.Second)
|
||||
lease.Release()
|
||||
return
|
||||
}
|
||||
dir := t.TempDir()
|
||||
ready := filepath.Join(dir, "ready")
|
||||
cmd := exec.Command(os.Args[0], "-test.run=TestLockReleasedAfterKill")
|
||||
cmd.Env = append(os.Environ(), "U8_GUARD_ROLE=hold", "U8_GUARD_DIR="+dir, "U8_GUARD_READY="+ready)
|
||||
cmd.Stdout = os.Stdout
|
||||
cmd.Stderr = os.Stderr
|
||||
if err := cmd.Start(); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
t.Cleanup(func() {
|
||||
if cmd.Process != nil {
|
||||
_ = cmd.Process.Kill()
|
||||
_, _ = cmd.Process.Wait()
|
||||
}
|
||||
})
|
||||
deadline := time.Now().Add(10 * time.Second)
|
||||
for time.Now().Before(deadline) {
|
||||
if _, err := os.Stat(ready); err == nil {
|
||||
break
|
||||
}
|
||||
time.Sleep(20 * time.Millisecond)
|
||||
}
|
||||
if _, err := os.Stat(ready); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if _, err := Acquire(dir); !errors.Is(err, ErrBusy) {
|
||||
t.Fatalf("held: %v", err)
|
||||
}
|
||||
if err := cmd.Process.Kill(); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
_, _ = cmd.Process.Wait()
|
||||
lease, err := Acquire(dir)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
lease.Release()
|
||||
}
|
||||
|
||||
func TestMissingDataDirIsCreatedBeforeLock(t *testing.T) {
|
||||
dir := filepath.Join(t.TempDir(), "nested", "data")
|
||||
lease, err := Acquire(dir)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
lease.Release()
|
||||
if _, err := os.Stat(dir); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
}
|
||||
|
||||
func TestLockSymlinkCannotAliasDatabase(t *testing.T) {
|
||||
dir := t.TempDir()
|
||||
db := filepath.Join(dir, "agent-system.db")
|
||||
st, err := store.OpenSQLite(db)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if err := st.Close(); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if err := os.Remove(db + ".lock"); err != nil && !os.IsNotExist(err) {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if err := os.Symlink(db, db+".lock"); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
before, err := os.ReadFile(db)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if _, err := Acquire(dir); !errors.Is(err, ErrLockAliased) {
|
||||
t.Fatalf("symlink lock: %v", err)
|
||||
}
|
||||
after, err := os.ReadFile(db)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if string(before) != string(after) {
|
||||
t.Fatal("aliased lock wrote into the database")
|
||||
}
|
||||
}
|
||||
|
||||
func TestSymlinkedDatabaseLocksResolvedFile(t *testing.T) {
|
||||
base := t.TempDir()
|
||||
realDir := filepath.Join(base, "real")
|
||||
aliasDir := filepath.Join(base, "alias")
|
||||
if err := os.Mkdir(realDir, 0o755); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if err := os.Mkdir(aliasDir, 0o755); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if err := os.WriteFile(filepath.Join(realDir, "agent-system.db"), []byte{}, 0o600); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if err := os.Symlink(filepath.Join(realDir, "agent-system.db"), filepath.Join(aliasDir, "agent-system.db")); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
held, err := Acquire(aliasDir)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
defer held.Release()
|
||||
if _, err := os.Stat(filepath.Join(realDir, "agent-system.db.lock")); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if _, err := Acquire(realDir); !errors.Is(err, ErrBusy) {
|
||||
t.Fatalf("resolved db lock: %v", err)
|
||||
}
|
||||
}
|
||||
|
||||
func openAndBind(t *testing.T, dir, instanceID string) (*store.Store, func(), error) {
|
||||
t.Helper()
|
||||
lease, err := Acquire(dir)
|
||||
if err != nil {
|
||||
return nil, nil, err
|
||||
}
|
||||
if err := lease.AutomaticRefusal(); err != nil {
|
||||
lease.Release()
|
||||
return nil, nil, err
|
||||
}
|
||||
st, err := store.OpenSQLite(lease.DBPath())
|
||||
if err != nil {
|
||||
lease.Release()
|
||||
return nil, nil, err
|
||||
}
|
||||
if err := Bind(st, lease, BindOptions{InstanceID: instanceID}); err != nil {
|
||||
_ = st.Close()
|
||||
lease.Release()
|
||||
return nil, nil, err
|
||||
}
|
||||
return st, func() {
|
||||
_ = st.Close()
|
||||
lease.Release()
|
||||
}, nil
|
||||
}
|
||||
@@ -0,0 +1,201 @@
|
||||
//go:build unix
|
||||
|
||||
package instanceguard
|
||||
|
||||
import (
|
||||
"fmt"
|
||||
"os"
|
||||
"path/filepath"
|
||||
"strings"
|
||||
"syscall"
|
||||
)
|
||||
|
||||
// Acquire 取得 SQLite 活动服务锁。失败时不返回 Lease。
|
||||
// 硬链接别名直接拒绝。锁文件被并发替换时重试数次后拒绝。
|
||||
func Acquire(dataDir string) (*Lease, error) {
|
||||
dir, err := normalizeDataDir(dataDir)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
if err := os.MkdirAll(dir, 0o755); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
_, dbPath, err := resolveDB(dir)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
if err := rejectHardlink(dbPath); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
lockPath := dbPath + ".lock"
|
||||
var last error
|
||||
for attempt := 0; attempt < 5; attempt++ {
|
||||
lease, retry, err := acquireOnce(dbPath, lockPath)
|
||||
if err == nil && !retry {
|
||||
return lease, nil
|
||||
}
|
||||
if retry {
|
||||
last = err
|
||||
continue
|
||||
}
|
||||
return nil, err
|
||||
}
|
||||
if last == nil {
|
||||
last = errorsRetry()
|
||||
}
|
||||
return nil, last
|
||||
}
|
||||
|
||||
func errorsRetry() error {
|
||||
return fmt.Errorf("instance guard: lock file inode changed while locking")
|
||||
}
|
||||
|
||||
func acquireOnce(dbPath, lockPath string) (*Lease, bool, error) {
|
||||
fd, err := syscall.Open(lockPath, syscall.O_RDWR|syscall.O_CREAT|syscall.O_CLOEXEC|syscall.O_NOFOLLOW, 0o600)
|
||||
if err != nil {
|
||||
if err == syscall.ELOOP {
|
||||
return nil, false, ErrLockAliased
|
||||
}
|
||||
return nil, false, err
|
||||
}
|
||||
fail := func(err error) (*Lease, bool, error) {
|
||||
syscall.Close(fd)
|
||||
return nil, false, err
|
||||
}
|
||||
if err := syscall.Flock(fd, syscall.LOCK_EX|syscall.LOCK_NB); err != nil {
|
||||
syscall.Close(fd)
|
||||
if err == syscall.EWOULDBLOCK || err == syscall.EAGAIN {
|
||||
return nil, false, ErrBusy
|
||||
}
|
||||
return nil, false, err
|
||||
}
|
||||
var held syscall.Stat_t
|
||||
if err := syscall.Fstat(fd, &held); err != nil {
|
||||
return fail(err)
|
||||
}
|
||||
var named syscall.Stat_t
|
||||
if err := syscall.Stat(lockPath, &named); err != nil {
|
||||
return fail(err)
|
||||
}
|
||||
if held.Ino != named.Ino || held.Dev != named.Dev {
|
||||
syscall.Flock(fd, syscall.LOCK_UN)
|
||||
syscall.Close(fd)
|
||||
return nil, true, errorsRetry()
|
||||
}
|
||||
buf := make([]byte, 512)
|
||||
n, err := syscall.Read(fd, buf)
|
||||
if err != nil {
|
||||
return fail(err)
|
||||
}
|
||||
rec, perr := parseRecord(string(buf[:n]))
|
||||
dbDev, dbIno, dbExists, err := statFile(dbPath)
|
||||
if err != nil {
|
||||
return fail(err)
|
||||
}
|
||||
if dbExists && dbDev == uint64(held.Dev) && dbIno == uint64(held.Ino) {
|
||||
syscall.Flock(fd, syscall.LOCK_UN)
|
||||
syscall.Close(fd)
|
||||
return nil, false, ErrLockAliased
|
||||
}
|
||||
lease := &Lease{
|
||||
fd: fd,
|
||||
dbPath: dbPath,
|
||||
lockDev: uint64(held.Dev),
|
||||
lockIno: uint64(held.Ino),
|
||||
dbDev: dbDev,
|
||||
dbIno: dbIno,
|
||||
dbExists: dbExists,
|
||||
record: rec,
|
||||
parseErr: perr,
|
||||
release: func() {
|
||||
syscall.Flock(fd, syscall.LOCK_UN)
|
||||
syscall.Close(fd)
|
||||
},
|
||||
}
|
||||
return lease, false, nil
|
||||
}
|
||||
|
||||
func (l *Lease) writeRecord(dev, ino uint64, instanceID, lockGen string) error {
|
||||
if strings.ContainsAny(instanceID, " \t\r\n") {
|
||||
return fmt.Errorf("instance guard: instance id contains whitespace")
|
||||
}
|
||||
line := fmt.Sprintf("v1\tdb_dev=%d\tdb_ino=%d\tinstance_id=%s\tlock_gen=%s\n", dev, ino, instanceID, lockGen)
|
||||
if _, err := syscall.Seek(l.fd, 0, 0); err != nil {
|
||||
return err
|
||||
}
|
||||
if _, err := syscall.Write(l.fd, []byte(line)); err != nil {
|
||||
return err
|
||||
}
|
||||
return syscall.Ftruncate(l.fd, int64(len(line)))
|
||||
}
|
||||
|
||||
func (l *Lease) currentDB() (dev, ino uint64, exists bool, err error) {
|
||||
return statFile(l.dbPath)
|
||||
}
|
||||
|
||||
func normalizeDataDir(dataDir string) (string, error) {
|
||||
if strings.TrimSpace(dataDir) == "" {
|
||||
dataDir = "."
|
||||
}
|
||||
abs, err := filepath.Abs(dataDir)
|
||||
if err != nil {
|
||||
return "", err
|
||||
}
|
||||
if _, err := os.Lstat(abs); err == nil {
|
||||
return filepath.EvalSymlinks(abs)
|
||||
} else if !os.IsNotExist(err) {
|
||||
return "", err
|
||||
}
|
||||
parent, base := filepath.Split(abs)
|
||||
parent = filepath.Clean(parent)
|
||||
if _, err := os.Lstat(parent); err == nil {
|
||||
resolved, err := filepath.EvalSymlinks(parent)
|
||||
if err != nil {
|
||||
return "", err
|
||||
}
|
||||
return filepath.Join(resolved, base), nil
|
||||
}
|
||||
return abs, nil
|
||||
}
|
||||
|
||||
func resolveDB(dataDir string) (logical, resolved string, err error) {
|
||||
logical = filepath.Join(dataDir, "agent-system.db")
|
||||
fi, err := os.Lstat(logical)
|
||||
if err != nil {
|
||||
if os.IsNotExist(err) {
|
||||
return logical, logical, nil
|
||||
}
|
||||
return "", "", err
|
||||
}
|
||||
if fi.Mode()&os.ModeSymlink != 0 {
|
||||
resolved, err = filepath.EvalSymlinks(logical)
|
||||
return logical, resolved, err
|
||||
}
|
||||
return logical, logical, nil
|
||||
}
|
||||
|
||||
func rejectHardlink(dbPath string) error {
|
||||
_, _, exists, err := statFile(dbPath)
|
||||
if err != nil || !exists {
|
||||
return err
|
||||
}
|
||||
var st syscall.Stat_t
|
||||
if err := syscall.Stat(dbPath, &st); err != nil {
|
||||
return err
|
||||
}
|
||||
if st.Nlink > 1 {
|
||||
return ErrHardlink
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
func statFile(path string) (dev, ino uint64, exists bool, err error) {
|
||||
var st syscall.Stat_t
|
||||
if err = syscall.Stat(path, &st); err != nil {
|
||||
if os.IsNotExist(err) {
|
||||
return 0, 0, false, nil
|
||||
}
|
||||
return 0, 0, false, err
|
||||
}
|
||||
return uint64(st.Dev), uint64(st.Ino), true, nil
|
||||
}
|
||||
@@ -4,6 +4,7 @@ import (
|
||||
"encoding/json"
|
||||
"net/http"
|
||||
"net/url"
|
||||
"strings"
|
||||
"testing"
|
||||
|
||||
"agent-system-backend/internal/store"
|
||||
@@ -113,7 +114,6 @@ func TestAccountHTTPPendingCreateProfileAndDisable(t *testing.T) {
|
||||
h := srv.Handler()
|
||||
adminCookie := bootAdmin(t, h, st)
|
||||
for _, body := range []string{
|
||||
`{"owner":"blocked-password","password":"old-pass-9"}`,
|
||||
`{"owner":"blocked-admin","isAdmin":true}`,
|
||||
} {
|
||||
if rec := do(h, "POST", "/api/admin/users", body, adminCookie); rec.Code != http.StatusBadRequest {
|
||||
@@ -152,3 +152,97 @@ func TestAccountHTTPPendingCreateProfileAndDisable(t *testing.T) {
|
||||
t.Fatalf("state/profile after disable: %+v %v", account, err)
|
||||
}
|
||||
}
|
||||
|
||||
func TestAdminPasswordCreateAndResetHTTP(t *testing.T) {
|
||||
srv, st := newAuthTestServer(t)
|
||||
h := srv.Handler()
|
||||
admin := bootAdmin(t, h, st)
|
||||
|
||||
created := do(h, "POST", "/api/admin/users", `{"owner":"alice","profile":{"name":"爱丽丝"},"password":"alice-initial-9"}`, admin)
|
||||
if created.Code != http.StatusOK {
|
||||
t.Fatalf("password create: %d %s", created.Code, created.Body.String())
|
||||
}
|
||||
if strings.Contains(created.Body.String(), "alice-initial-9") {
|
||||
t.Fatal("supplied password was echoed")
|
||||
}
|
||||
var createdBody struct {
|
||||
UserID string `json:"userId"`
|
||||
PasswordSet bool `json:"passwordSet"`
|
||||
MustChangePassword bool `json:"mustChangePassword"`
|
||||
InitialPassword string `json:"initialPassword"`
|
||||
}
|
||||
if err := json.Unmarshal(created.Body.Bytes(), &createdBody); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if createdBody.UserID == "" || !createdBody.PasswordSet || !createdBody.MustChangePassword || createdBody.InitialPassword != "" {
|
||||
t.Fatalf("supplied password projection: %+v", createdBody)
|
||||
}
|
||||
account, err := st.GetAccountByOwner("alice")
|
||||
if err != nil || !account.HasPassword || !account.MustChangePassword || account.InitializationState != store.InitializationInitialized {
|
||||
t.Fatalf("stored account state: %+v %v", account, err)
|
||||
}
|
||||
if state, ok := st.VerifyLogin("alice", "alice-initial-9"); !ok || !state.MustChangePassword {
|
||||
t.Fatalf("supplied password login: ok=%v state=%+v", ok, state)
|
||||
}
|
||||
|
||||
generated := do(h, "POST", "/api/admin/users", `{"owner":"bob"}`, admin)
|
||||
if generated.Code != http.StatusOK {
|
||||
t.Fatalf("generated password create: %d %s", generated.Code, generated.Body.String())
|
||||
}
|
||||
var generatedBody struct {
|
||||
InitialPassword string `json:"initialPassword"`
|
||||
MustChangePassword bool `json:"mustChangePassword"`
|
||||
}
|
||||
if err := json.Unmarshal(generated.Body.Bytes(), &generatedBody); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if generatedBody.InitialPassword == "" || !generatedBody.MustChangePassword {
|
||||
t.Fatalf("generated password projection: %+v", generatedBody)
|
||||
}
|
||||
if state, ok := st.VerifyLogin("bob", generatedBody.InitialPassword); !ok || !state.MustChangePassword {
|
||||
t.Fatalf("generated password login: ok=%v state=%+v", ok, state)
|
||||
}
|
||||
|
||||
login := do(h, "POST", "/api/auth/login", `{"owner":"alice","password":"alice-initial-9"}`, "")
|
||||
if login.Code != http.StatusOK {
|
||||
t.Fatalf("alice login: %d %s", login.Code, login.Body.String())
|
||||
}
|
||||
aliceSession := cookieOf(login)
|
||||
key, _, err := st.IssueLoginKey("alice", store.LoginKeyPurposeRecover, "test", 0)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if rejected := do(h, "POST", "/api/admin/users/alice/password", `{"password":"client-supplied"}`, admin); rejected.Code != http.StatusBadRequest {
|
||||
t.Fatalf("client reset password accepted: %d %s", rejected.Code, rejected.Body.String())
|
||||
}
|
||||
if _, ok := st.VerifyLogin("alice", "alice-initial-9"); !ok {
|
||||
t.Fatal("rejected reset changed the password")
|
||||
}
|
||||
|
||||
reset := do(h, "POST", "/api/admin/users/alice/password", `{}`, admin)
|
||||
if reset.Code != http.StatusOK || reset.Header().Get("Cache-Control") != "no-store" {
|
||||
t.Fatalf("admin reset: %d %s", reset.Code, reset.Body.String())
|
||||
}
|
||||
var resetBody struct {
|
||||
TemporaryPassword string `json:"temporaryPassword"`
|
||||
MustChangePassword bool `json:"mustChangePassword"`
|
||||
}
|
||||
if err := json.Unmarshal(reset.Body.Bytes(), &resetBody); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if resetBody.TemporaryPassword == "" || resetBody.TemporaryPassword == "alice-initial-9" || !resetBody.MustChangePassword {
|
||||
t.Fatalf("reset projection: %+v", resetBody)
|
||||
}
|
||||
if rec := do(h, "GET", "/api/me", "", aliceSession); rec.Code != http.StatusUnauthorized {
|
||||
t.Fatalf("reset did not revoke session: %d %s", rec.Code, rec.Body.String())
|
||||
}
|
||||
if _, ok := st.VerifyLogin("alice", "alice-initial-9"); ok {
|
||||
t.Fatal("old password remained usable after reset")
|
||||
}
|
||||
if state, ok := st.VerifyLogin("alice", resetBody.TemporaryPassword); !ok || !state.MustChangePassword {
|
||||
t.Fatalf("temporary password login: ok=%v state=%+v", ok, state)
|
||||
}
|
||||
if _, _, err := st.RedeemLoginKey(key); err == nil {
|
||||
t.Fatal("admin reset left unused login key redeemable")
|
||||
}
|
||||
}
|
||||
|
||||
+50
-10
@@ -324,8 +324,9 @@ func (s *Server) createAdminUser(w http.ResponseWriter, r *http.Request) {
|
||||
return
|
||||
}
|
||||
var req struct {
|
||||
Owner string `json:"owner"`
|
||||
Profile store.AccountProfile `json:"profile"`
|
||||
Owner string `json:"owner"`
|
||||
Profile store.AccountProfile `json:"profile"`
|
||||
Password *string `json:"password"`
|
||||
}
|
||||
if err := decodeStrictJSON(r, &req); err != nil {
|
||||
http.Error(w, err.Error(), http.StatusBadRequest)
|
||||
@@ -336,12 +337,31 @@ func (s *Server) createAdminUser(w http.ResponseWriter, r *http.Request) {
|
||||
http.Error(w, "owner required (no spaces or slashes)", http.StatusBadRequest)
|
||||
return
|
||||
}
|
||||
account, err := s.st.CreatePendingAccount(actor, req.Owner, req.Profile)
|
||||
password := RandomPassword()
|
||||
generated := req.Password == nil
|
||||
if req.Password != nil {
|
||||
password = *req.Password
|
||||
if err := validPassword(password); err != nil {
|
||||
http.Error(w, err.Error(), http.StatusBadRequest)
|
||||
return
|
||||
}
|
||||
}
|
||||
hash, err := store.HashPassword(password)
|
||||
if err != nil {
|
||||
http.Error(w, err.Error(), http.StatusInternalServerError)
|
||||
return
|
||||
}
|
||||
account, err := s.st.CreateAdminPasswordAccount(actor, req.Owner, req.Profile, hash)
|
||||
if err != nil {
|
||||
writeStoreErr(w, err)
|
||||
return
|
||||
}
|
||||
writeJSON(w, http.StatusOK, accountProfileJSON(account))
|
||||
out := accountProfileJSON(account)
|
||||
if generated {
|
||||
w.Header().Set("Cache-Control", "no-store")
|
||||
out["initialPassword"] = password
|
||||
}
|
||||
writeJSON(w, http.StatusOK, out)
|
||||
}
|
||||
|
||||
func accountProfileJSON(account store.Account) map[string]any {
|
||||
@@ -349,7 +369,8 @@ func accountProfileJSON(account store.Account) map[string]any {
|
||||
"userId": account.ID, "owner": account.Owner, "profile": account.Profile,
|
||||
"profileVersion": account.ProfileVersion, "enabled": account.Enabled,
|
||||
"stateVersion": account.StateVersion, "isAdmin": account.IsAdmin,
|
||||
"passwordSet": account.HasPassword, "initializationState": account.InitializationState,
|
||||
"passwordSet": account.HasPassword, "mustChangePassword": account.MustChangePassword,
|
||||
"initializationState": account.InitializationState,
|
||||
"initializationGeneration": account.InitializationGeneration,
|
||||
}
|
||||
}
|
||||
@@ -462,13 +483,32 @@ func (s *Server) setAdminAccountState(w http.ResponseWriter, r *http.Request) {
|
||||
writeJSON(w, http.StatusOK, accountProfileJSON(account))
|
||||
}
|
||||
|
||||
// resetAdminUserPassword:管理员直接重置密码(兼容保留的既有入口;常规恢复
|
||||
// 走签发一次性登录 Key)。单事务:must_change=1、bump auth_version、作废未用
|
||||
// 登录 Key、吊销全部会话、审计。
|
||||
// resetAdminUserPassword:管理员生成随机临时密码。单事务:must_change=1、bump
|
||||
// auth_version、作废未用登录 Key、吊销全部会话、审计。
|
||||
func (s *Server) resetAdminUserPassword(w http.ResponseWriter, r *http.Request) {
|
||||
if _, ok := s.requireAdmin(w, r); ok {
|
||||
http.Error(w, "RECOVERY_CONTRACT_REPLACED: propose recovery instead", http.StatusGone)
|
||||
actor, ok := s.requireAdmin(w, r)
|
||||
if !ok {
|
||||
return
|
||||
}
|
||||
var req struct{}
|
||||
if err := decodeStrictJSON(r, &req); err != nil {
|
||||
http.Error(w, err.Error(), http.StatusBadRequest)
|
||||
return
|
||||
}
|
||||
password := RandomPassword()
|
||||
hash, err := store.HashPassword(password)
|
||||
if err != nil {
|
||||
http.Error(w, err.Error(), http.StatusInternalServerError)
|
||||
return
|
||||
}
|
||||
if err := s.st.AdminResetUserPassword(r.PathValue("owner"), hash, actor, "admin-http"); err != nil {
|
||||
writeStoreErr(w, err)
|
||||
return
|
||||
}
|
||||
w.Header().Set("Cache-Control", "no-store")
|
||||
writeJSON(w, http.StatusOK, map[string]any{
|
||||
"owner": r.PathValue("owner"), "temporaryPassword": password, "mustChangePassword": true,
|
||||
})
|
||||
}
|
||||
|
||||
// ---- 管理员:一次性登录 Key(恢复流程主入口)----
|
||||
|
||||
@@ -423,19 +423,33 @@ func TestAdminAccountFlow(t *testing.T) {
|
||||
t.Fatalf("non-admin create must 403: %d", rec.Code)
|
||||
}
|
||||
|
||||
// 旧管理员重置入口关闭,不能动 alice 的密码与当前会话。
|
||||
if rec := session("POST", "/api/admin/users/alice/password", `{"password":"reset-init-02"}`); rec.Code != http.StatusGone {
|
||||
t.Fatalf("admin reset must be closed: %d %s", rec.Code, rec.Body.String())
|
||||
// 管理员重置总是生成临时密码,销毁 alice 的旧会话并强制首登改密。
|
||||
reset := session("POST", "/api/admin/users/alice/password", `{}`)
|
||||
if reset.Code != http.StatusOK || reset.Header().Get("Cache-Control") != "no-store" {
|
||||
t.Fatalf("admin reset: %d %s", reset.Code, reset.Body.String())
|
||||
}
|
||||
if rec := alice("GET", "/api/me", ""); rec.Code != http.StatusOK {
|
||||
t.Fatalf("closed reset must preserve alice session: %d", rec.Code)
|
||||
var resetBody struct {
|
||||
TemporaryPassword string `json:"temporaryPassword"`
|
||||
}
|
||||
if err := json.Unmarshal(reset.Body.Bytes(), &resetBody); err != nil || resetBody.TemporaryPassword == "" {
|
||||
t.Fatalf("admin reset response: %+v %v", resetBody, err)
|
||||
}
|
||||
if rec := alice("GET", "/api/me", ""); rec.Code != http.StatusUnauthorized {
|
||||
t.Fatalf("reset must revoke alice session: %d", rec.Code)
|
||||
}
|
||||
req = httptest.NewRequest("POST", "/api/auth/login", strings.NewReader(`{"owner":"alice","password":"alice-final-9"}`))
|
||||
req.Header.Set("Content-Type", "application/json")
|
||||
rec = httptest.NewRecorder()
|
||||
h.ServeHTTP(rec, req)
|
||||
if rec.Code != http.StatusOK || strings.Contains(rec.Body.String(), `"mustChangePassword":true`) {
|
||||
t.Fatalf("closed reset must preserve normal login: %d %s", rec.Code, rec.Body.String())
|
||||
if rec.Code != http.StatusUnauthorized {
|
||||
t.Fatalf("old password remained usable after reset: %d %s", rec.Code, rec.Body.String())
|
||||
}
|
||||
req = httptest.NewRequest("POST", "/api/auth/login", strings.NewReader(`{"owner":"alice","password":"`+resetBody.TemporaryPassword+`"}`))
|
||||
req.Header.Set("Content-Type", "application/json")
|
||||
rec = httptest.NewRecorder()
|
||||
h.ServeHTTP(rec, req)
|
||||
if rec.Code != http.StatusOK || !strings.Contains(rec.Body.String(), `"mustChangePassword":true`) {
|
||||
t.Fatalf("temporary password login: %d %s", rec.Code, rec.Body.String())
|
||||
}
|
||||
|
||||
// API key 通道兼容:密钥登录照旧(并具备其 owner 身份)。
|
||||
|
||||
@@ -99,7 +99,7 @@ func TestOwnerConfirmedRecoveryHTTPKeepsGrantFromAdministrator(t *testing.T) {
|
||||
|
||||
func jsonNumber(v int64) string { return strconv.FormatInt(v, 10) }
|
||||
|
||||
func TestLegacyAdministratorRecoveryEntrancesAreClosed(t *testing.T) {
|
||||
func TestLegacyAdministratorLoginKeyIssuanceEntrancesAreClosed(t *testing.T) {
|
||||
srv, st := newAuthTestServer(t)
|
||||
h := srv.Handler()
|
||||
admin := bootAdmin(t, h, st)
|
||||
@@ -114,7 +114,6 @@ func TestLegacyAdministratorRecoveryEntrancesAreClosed(t *testing.T) {
|
||||
t.Fatal(err)
|
||||
}
|
||||
for _, route := range []struct{ path, body string }{
|
||||
{"/api/admin/users/victim/password", `{"password":"attacker-pass"}`},
|
||||
{"/api/admin/users/victim/login-keys", `{"purpose":"recover"}`},
|
||||
{"/api/admin/users/victim/login-keys", `{"purpose":"activate"}`},
|
||||
} {
|
||||
|
||||
@@ -299,6 +299,53 @@ func (s *Store) CreatePendingAccount(actorOwner, owner string, profile AccountPr
|
||||
return a, nil
|
||||
}
|
||||
|
||||
// CreateAdminPasswordAccount creates an ordinary user with an administrator-
|
||||
// supplied password hash. The account row, hash, and audit entries commit
|
||||
// together so a failed write cannot leave a credential without its account.
|
||||
func (s *Store) CreateAdminPasswordAccount(actorOwner, owner string, profile AccountProfile, passwordHash string) (Account, error) {
|
||||
if passwordHash == "" {
|
||||
return Account{}, ErrConflict
|
||||
}
|
||||
tx, err := s.db.Begin()
|
||||
if err != nil {
|
||||
return Account{}, err
|
||||
}
|
||||
defer tx.Rollback()
|
||||
locked, err := s.lockAccountUsers(tx, actorOwner)
|
||||
if errors.Is(err, ErrNotFound) {
|
||||
return Account{}, ErrAccountForbidden
|
||||
}
|
||||
if err != nil {
|
||||
return Account{}, err
|
||||
}
|
||||
actor := locked[actorOwner]
|
||||
if !actor.Enabled || !actor.IsAdmin {
|
||||
return Account{}, ErrAccountForbidden
|
||||
}
|
||||
a, err := s.insertPendingAccountTx(tx, owner, profile)
|
||||
if err != nil {
|
||||
return Account{}, err
|
||||
}
|
||||
if _, err := tx.Exec(s.q(`UPDATE users SET password_hash=?,password_set_at=?,must_change_password=?,credential_origin=?,initialization_state=? WHERE id=?`),
|
||||
passwordHash, nowISO(), true, CredentialOriginAdminPassword, InitializationInitialized, a.ID); err != nil {
|
||||
return Account{}, err
|
||||
}
|
||||
a, err = s.accountByOwner(tx, owner)
|
||||
if err != nil {
|
||||
return Account{}, err
|
||||
}
|
||||
if err := s.insertAccountEvent(tx, actorOwner, "account-management", owner, "account_created_password", a.ID); err != nil {
|
||||
return Account{}, err
|
||||
}
|
||||
if err := s.auditAccountMutation(tx, owner, actorOwner, "account.create_password"); err != nil {
|
||||
return Account{}, err
|
||||
}
|
||||
if err := tx.Commit(); err != nil {
|
||||
return Account{}, err
|
||||
}
|
||||
return a, nil
|
||||
}
|
||||
|
||||
// insertPendingAccountTx lets the trusted local initializer create and grant in
|
||||
// one transaction without impersonating an administrator. The caller owns its
|
||||
// authorization, audit and commit; this helper always inserts an ordinary user.
|
||||
|
||||
@@ -0,0 +1,67 @@
|
||||
package store
|
||||
|
||||
import (
|
||||
"fmt"
|
||||
"strings"
|
||||
)
|
||||
|
||||
// DurabilityRefusal 判断 PostgreSQL 耐久参数。off 或缺失即拒绝。
|
||||
// synchronous_commit=local 不是 off,不拒绝。
|
||||
func DurabilityRefusal(settings map[string]string) error {
|
||||
for _, key := range []string{"fsync", "full_page_writes", "synchronous_commit"} {
|
||||
value, ok := settings[key]
|
||||
if !ok || strings.EqualFold(strings.TrimSpace(value), "off") {
|
||||
return fmt.Errorf("durability: %s=%q", key, value)
|
||||
}
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
func (s *Store) verifyDurability() error {
|
||||
switch s.dialect.name() {
|
||||
case "sqlite":
|
||||
return s.verifySQLiteDurability()
|
||||
case "postgres":
|
||||
return s.verifyPostgresDurability()
|
||||
default:
|
||||
return fmt.Errorf("durability: unknown dialect %s", s.dialect.name())
|
||||
}
|
||||
}
|
||||
|
||||
func (s *Store) verifySQLiteDurability() error {
|
||||
var journal string
|
||||
if err := s.db.QueryRow(`PRAGMA journal_mode`).Scan(&journal); err != nil {
|
||||
return fmt.Errorf("durability: journal_mode: %w", err)
|
||||
}
|
||||
var syncLevel int64
|
||||
if err := s.db.QueryRow(`PRAGMA synchronous`).Scan(&syncLevel); err != nil {
|
||||
return fmt.Errorf("durability: synchronous: %w", err)
|
||||
}
|
||||
if !strings.EqualFold(journal, "wal") || syncLevel != 2 {
|
||||
return fmt.Errorf("durability: journal_mode=%s synchronous=%d", journal, syncLevel)
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
func (s *Store) verifyPostgresDurability() error {
|
||||
rows, err := s.db.Query(`SELECT name, setting FROM pg_settings WHERE name IN ('fsync','full_page_writes','synchronous_commit')`)
|
||||
if err != nil {
|
||||
return fmt.Errorf("durability: pg_settings: %w", err)
|
||||
}
|
||||
defer rows.Close()
|
||||
got := map[string]string{}
|
||||
for rows.Next() {
|
||||
var name, setting string
|
||||
if err := rows.Scan(&name, &setting); err != nil {
|
||||
return err
|
||||
}
|
||||
got[name] = setting
|
||||
}
|
||||
if err := rows.Err(); err != nil {
|
||||
return err
|
||||
}
|
||||
if err := DurabilityRefusal(got); err != nil {
|
||||
return err
|
||||
}
|
||||
return nil
|
||||
}
|
||||
@@ -0,0 +1,95 @@
|
||||
package store
|
||||
|
||||
import (
|
||||
"os"
|
||||
"path/filepath"
|
||||
"strings"
|
||||
"testing"
|
||||
|
||||
core "agent-system-core"
|
||||
)
|
||||
|
||||
func TestDurabilityRefusal(t *testing.T) {
|
||||
ok := map[string]string{
|
||||
"fsync": "on", "full_page_writes": "on", "synchronous_commit": "local",
|
||||
}
|
||||
if err := DurabilityRefusal(ok); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
for _, key := range []string{"fsync", "full_page_writes", "synchronous_commit"} {
|
||||
bad := map[string]string{
|
||||
"fsync": "on", "full_page_writes": "on", "synchronous_commit": "on",
|
||||
}
|
||||
bad[key] = "off"
|
||||
if err := DurabilityRefusal(bad); err == nil {
|
||||
t.Fatalf("%s off should refuse", key)
|
||||
}
|
||||
delete(bad, key)
|
||||
if err := DurabilityRefusal(bad); err == nil {
|
||||
t.Fatalf("missing %s should refuse", key)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
func TestSQLitePragmaReadback(t *testing.T) {
|
||||
st, err := OpenSQLite(filepath.Join(t.TempDir(), "agent-system.db"))
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
defer st.Close()
|
||||
var journal string
|
||||
if err := st.db.QueryRow(`PRAGMA journal_mode`).Scan(&journal); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
var syncLevel int64
|
||||
if err := st.db.QueryRow(`PRAGMA synchronous`).Scan(&syncLevel); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if !strings.EqualFold(journal, "wal") || syncLevel != 2 {
|
||||
t.Fatalf("journal=%s synchronous=%d", journal, syncLevel)
|
||||
}
|
||||
marker, ok, err := st.LoadInstanceGuard()
|
||||
if err != nil || ok || marker.InstanceID != "" {
|
||||
t.Fatalf("fresh db has no guard row: ok=%v err=%v", ok, err)
|
||||
}
|
||||
}
|
||||
|
||||
func TestOpenExistingDoesNotCreateMissingParent(t *testing.T) {
|
||||
missing := filepath.Join(t.TempDir(), "absent", "agent-system.db")
|
||||
if _, err := OpenExistingSQLite(missing); err == nil {
|
||||
t.Fatal("missing file opened")
|
||||
}
|
||||
if _, err := os.Stat(filepath.Dir(missing)); !os.IsNotExist(err) {
|
||||
t.Fatalf("parent created: %v", err)
|
||||
}
|
||||
}
|
||||
|
||||
func TestRunSurvivesReopen(t *testing.T) {
|
||||
path := filepath.Join(t.TempDir(), "agent-system.db")
|
||||
st, err := OpenSQLite(path)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
run := &core.Run{
|
||||
ID: "run_keep", ConversationID: "c", ProjectID: "p", Owner: "o",
|
||||
Kind: core.RunConversation, Status: core.RunWaitingHITL,
|
||||
}
|
||||
if err := st.CreateRun(run); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if err := st.Close(); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
st, err = OpenSQLite(path)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
defer st.Close()
|
||||
got, err := st.GetRunByID("run_keep")
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if got.Status != core.RunWaitingHITL {
|
||||
t.Fatalf("status %s", got.Status)
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,35 @@
|
||||
//go:build unix
|
||||
|
||||
package store
|
||||
|
||||
import (
|
||||
"os/signal"
|
||||
"path/filepath"
|
||||
"strings"
|
||||
"syscall"
|
||||
"testing"
|
||||
)
|
||||
|
||||
func TestWriteFailureIsReturned(t *testing.T) {
|
||||
st, err := OpenSQLite(filepath.Join(t.TempDir(), "agent-system.db"))
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
defer st.Close()
|
||||
signal.Ignore(syscall.SIGXFSZ)
|
||||
t.Cleanup(func() { signal.Reset(syscall.SIGXFSZ) })
|
||||
var lim syscall.Rlimit
|
||||
if err := syscall.Getrlimit(syscall.RLIMIT_FSIZE, &lim); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
old := lim
|
||||
lim.Cur = 4096
|
||||
if err := syscall.Setrlimit(syscall.RLIMIT_FSIZE, &lim); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
t.Cleanup(func() { _ = syscall.Setrlimit(syscall.RLIMIT_FSIZE, &old) })
|
||||
_, err = st.CreateProject("owner", strings.Repeat("x", 64*1024))
|
||||
if err == nil {
|
||||
t.Fatal("oversized write returned success")
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,47 @@
|
||||
package store
|
||||
|
||||
import (
|
||||
"database/sql"
|
||||
"errors"
|
||||
)
|
||||
|
||||
// InstanceGuardMarker 是 instance_guard 表的唯一一行。
|
||||
type InstanceGuardMarker struct {
|
||||
DBDev uint64
|
||||
DBIno uint64
|
||||
LockDev uint64
|
||||
LockIno uint64
|
||||
InstanceID string
|
||||
LockGen string
|
||||
}
|
||||
|
||||
// LoadInstanceGuard 读取守卫标记。没有行时 ok 为 false。
|
||||
func (s *Store) LoadInstanceGuard() (InstanceGuardMarker, bool, error) {
|
||||
var m InstanceGuardMarker
|
||||
var dbDev, dbIno, lockDev, lockIno int64
|
||||
err := s.db.QueryRow(`SELECT db_dev, db_ino, lock_dev, lock_ino, instance_id, lock_gen FROM instance_guard WHERE id=1`).
|
||||
Scan(&dbDev, &dbIno, &lockDev, &lockIno, &m.InstanceID, &m.LockGen)
|
||||
if errors.Is(err, sql.ErrNoRows) {
|
||||
return InstanceGuardMarker{}, false, nil
|
||||
}
|
||||
if err != nil {
|
||||
return InstanceGuardMarker{}, false, err
|
||||
}
|
||||
m.DBDev = uint64(dbDev)
|
||||
m.DBIno = uint64(dbIno)
|
||||
m.LockDev = uint64(lockDev)
|
||||
m.LockIno = uint64(lockIno)
|
||||
return m, true, nil
|
||||
}
|
||||
|
||||
// SaveInstanceGuard 写入或覆盖唯一一行。
|
||||
func (s *Store) SaveInstanceGuard(m InstanceGuardMarker) error {
|
||||
_, err := s.db.Exec(s.q(`INSERT INTO instance_guard(id, db_dev, db_ino, lock_dev, lock_ino, instance_id, lock_gen)
|
||||
VALUES(1,?,?,?,?,?,?)
|
||||
ON CONFLICT(id) DO UPDATE SET
|
||||
db_dev=excluded.db_dev, db_ino=excluded.db_ino,
|
||||
lock_dev=excluded.lock_dev, lock_ino=excluded.lock_ino,
|
||||
instance_id=excluded.instance_id, lock_gen=excluded.lock_gen`),
|
||||
int64(m.DBDev), int64(m.DBIno), int64(m.LockDev), int64(m.LockIno), m.InstanceID, m.LockGen)
|
||||
return err
|
||||
}
|
||||
@@ -901,6 +901,34 @@ func init() {
|
||||
up: map[string][]string{"sqlite": execEnvUp, "postgres": execEnvUp},
|
||||
down: map[string][]string{"sqlite": execEnvDown, "postgres": execEnvDown},
|
||||
},
|
||||
// U8:活动服务锁的库内标记。锁文件 inode 与本行不一致表示锁文件被重建。
|
||||
migration{version: 36, name: "instance_guard",
|
||||
up: map[string][]string{"sqlite": {
|
||||
`CREATE TABLE IF NOT EXISTS instance_guard (
|
||||
id INTEGER PRIMARY KEY CHECK (id = 1),
|
||||
db_dev INTEGER NOT NULL,
|
||||
db_ino INTEGER NOT NULL,
|
||||
lock_dev INTEGER NOT NULL,
|
||||
lock_ino INTEGER NOT NULL,
|
||||
instance_id TEXT NOT NULL DEFAULT '',
|
||||
lock_gen TEXT NOT NULL DEFAULT ''
|
||||
)`,
|
||||
}, "postgres": {
|
||||
`CREATE TABLE IF NOT EXISTS instance_guard (
|
||||
id INTEGER PRIMARY KEY CHECK (id = 1),
|
||||
db_dev BIGINT NOT NULL,
|
||||
db_ino BIGINT NOT NULL,
|
||||
lock_dev BIGINT NOT NULL,
|
||||
lock_ino BIGINT NOT NULL,
|
||||
instance_id TEXT NOT NULL DEFAULT '',
|
||||
lock_gen TEXT NOT NULL DEFAULT ''
|
||||
)`,
|
||||
}},
|
||||
down: map[string][]string{"sqlite": {
|
||||
`DROP TABLE IF EXISTS instance_guard`,
|
||||
}, "postgres": {
|
||||
`DROP TABLE IF EXISTS instance_guard`,
|
||||
}}},
|
||||
)
|
||||
}
|
||||
|
||||
|
||||
@@ -40,6 +40,7 @@ var migrationBreaksOldReaders = map[int]bool{
|
||||
33: true, // 本人证明与 grant 绑定协议;旧恢复写者和兑换者必须排除
|
||||
34: false, // 游标签名密钥只新增内部列;旧读者按列名读取既有字段
|
||||
35: false, // exec_env 新表;旧读者不读这些表
|
||||
36: false, // instance_guard 是新表,旧制品不读取它
|
||||
}
|
||||
|
||||
// SchemaTarget 是本二进制会迁移到的 schema 版本。
|
||||
|
||||
@@ -189,7 +189,8 @@ func openSQLiteFile(path string, createArtifacts, migrateSchema bool) (*Store, e
|
||||
// 跨进程迁移/写路径依赖 busy_timeout 排队而不是中途 SQLITE_BUSY(KTD8)。
|
||||
// busy_timeout 30s:迁移竞争测试在慢速 CI 盘上 4 个迁移者串行全程可超 5s
|
||||
// (实测 job 2285 SQLITE_BUSY);本地档宁可排队不报错。
|
||||
dsn := fmt.Sprintf("file:%s?_pragma=busy_timeout(30000)&_pragma=journal_mode(WAL)&_pragma=foreign_keys(1)&_txlock=immediate", path)
|
||||
// synchronous(FULL) 是恢复方案 KTD14 的可靠档。打开后在同一连接读回,见 verifyDurability。
|
||||
dsn := fmt.Sprintf("file:%s?_pragma=busy_timeout(30000)&_pragma=journal_mode(WAL)&_pragma=synchronous(FULL)&_pragma=foreign_keys(1)&_txlock=immediate", path)
|
||||
db, err := sql.Open("sqlite", dsn)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
@@ -293,6 +294,10 @@ func finishOpenMode(db *sql.DB, d dialect, artifactsDir string, blobs blobstore.
|
||||
db.Close()
|
||||
return nil, err
|
||||
}
|
||||
if err := s.verifyDurability(); err != nil {
|
||||
db.Close()
|
||||
return nil, err
|
||||
}
|
||||
if createArtifacts {
|
||||
if err := os.MkdirAll(artifactsDir, 0o755); err != nil {
|
||||
db.Close()
|
||||
|
||||
Reference in New Issue
Block a user