diff --git a/cmd/agent-system/main.go b/cmd/agent-system/main.go index 391d27a..c9d4f14 100644 --- a/cmd/agent-system/main.go +++ b/cmd/agent-system/main.go @@ -35,6 +35,8 @@ func main() { os.Exit(runUser(os.Args[2:])) case "identity": os.Exit(runIdentity(os.Args[2:])) + case "migrate": + os.Exit(runMigrate(os.Args[2:])) } } fs := flag.NewFlagSet("serve", flag.ExitOnError) @@ -59,8 +61,10 @@ 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 活动服务守卫") fs.Parse(os.Args[1:]) 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, @@ -85,8 +89,9 @@ func main() { StorageDSN: resolved.StorageDSN, ArtifactsDir: resolved.ArtifactsDir, AuthMode: resolved.AuthMode, InstanceID: resolved.Instance, InsecureListen: resolved.InsecureListen, AllowedHosts: resolved.AllowedHosts(), DisableExecTools: resolved.DisableExecTools, ExecBackend: resolved.ExecBackend, ExecDepsDir: resolved.ExecDepsDir, - ExecWorkspace: resolved.ExecWorkspace, - Tier: resolved.Tier, + ExecWorkspace: resolved.ExecWorkspace, + Tier: resolved.Tier, + AdoptInstanceGuard: *adoptGuard, PGPool: store.PoolOptions{ MaxOpenConn: resolved.PGMaxOpen, MaxIdleConn: resolved.PGMaxIdle, ConnMaxIdleTimeSec: resolved.PGConnMaxIdleSec, ConnMaxLifetimeSec: resolved.PGConnMaxLifetimeSec, diff --git a/cmd/agent-system/migrate.go b/cmd/agent-system/migrate.go new file mode 100644 index 0000000..21dd629 --- /dev/null +++ b/cmd/agent-system/migrate.go @@ -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 +} diff --git a/internal/app/app.go b/internal/app/app.go index 2244da4..4d050f6 100644 --- a/internal/app/app.go +++ b/internal/app/app.go @@ -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" @@ -93,6 +92,8 @@ type Options struct { Tier string // PGPool PG 连接预算（KTD12；零值走默认）。仅 PG 档生效。 PGPool store.PoolOptions + // AdoptInstanceGuard 在数据库被替换或锁文件被重建后，显式重写守卫记录。 + AdoptInstanceGuard bool } // instanceTag 稳定执行身份：tier=team 时强制 host:addr（多实例互异）， @@ -171,34 +172,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) + return err } - defer st.Close() + defer releaseStore() // U1 I3：svc 注入的服务实例身份（manifest.instance_id）与库内 marker 对齐； // 未设置则跳过（兼容未接线旧调用方）。与 --instance（runs 进程标识）无关。 if sysID := strings.TrimSpace(os.Getenv("AGENT_SYSTEM_INSTANCE_ID")); sysID != "" { diff --git a/internal/app/guard_test.go b/internal/app/guard_test.go new file mode 100644 index 0000000..f147c06 --- /dev/null +++ b/internal/app/guard_test.go @@ -0,0 +1,151 @@ +//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") +} diff --git a/internal/app/service_store.go b/internal/app/service_store.go new file mode 100644 index 0000000..2494a4f --- /dev/null +++ b/internal/app/service_store.go @@ -0,0 +1,80 @@ +package app + +import ( + "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") + } + 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" { + return nil, nop, fmt.Errorf("team 档打开 PostgreSQL 失败，拒绝就绪（不回落 SQLite）: %w", err) + } + if err != nil { + return nil, nop, fmt.Errorf("open store: %w", 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 !opt.AdoptInstanceGuard { + if err := lease.AutomaticRefusal(); err != nil { + 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 := 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 +} + +// Migrate 执行显式迁移。SQLite 必须先取得活动服务锁，有服务占用则拒绝。 +func Migrate(opt Options) error { + _, release, err := openServiceStore(opt) + if err != nil { + return err + } + release() + return nil +} diff --git a/internal/instanceguard/guard.go b/internal/instanceguard/guard.go new file mode 100644 index 0000000..06a28f2 --- /dev/null +++ b/internal/instanceguard/guard.go @@ -0,0 +1,148 @@ +// Package instanceguard 强制同一 SQLite 数据文件只有一个活动服务。 +// 锁在打开数据库（迁移发生在打开内部）之前取得。账户 CLI 不使用本包。 +package instanceguard + +import ( + "errors" + "fmt" + "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") +) + +// Record 是锁文件里记下的数据库身份。没有记录表示锁文件是新建的。 +type Record struct { + DBDev uint64 + DBIno uint64 + InstanceID 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 && (marker.LockDev != lease.lockDev || marker.LockIno != lease.lockIno) && !opt.Adopt { + 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 + next := store.InstanceGuardMarker{ + DBDev: dev, DBIno: ino, LockDev: lease.lockDev, LockIno: lease.lockIno, InstanceID: id, + } + if err := lease.writeRecord(dev, ino, id); err != nil { + return err + } + if !ok || opt.Adopt || marker != next { + if err := st.SaveInstanceGuard(next); err != nil { + return err + } + } + return nil +} + +func parseRecord(raw string) (*Record, error) { + raw = strings.TrimSpace(raw) + if raw == "" { + return nil, nil + } + var dev, ino uint64 + var id string + if _, err := fmt.Sscanf(raw, "v1\tdb_dev=%d\tdb_ino=%d\tinstance_id=%s", &dev, &ino, &id); err == nil { + want := fmt.Sprintf("v1\tdb_dev=%d\tdb_ino=%d\tinstance_id=%s", dev, ino, id) + if raw == want && !strings.ContainsAny(id, " \t") { + return &Record{DBDev: dev, DBIno: ino, InstanceID: id}, nil + } + } + if _, err := fmt.Sscanf(raw, "v1\tdb_dev=%d\tdb_ino=%d\tinstance_id=", &dev, &ino); err == nil { + want := fmt.Sprintf("v1\tdb_dev=%d\tdb_ino=%d\tinstance_id=", dev, ino) + if raw == want { + return &Record{DBDev: dev, DBIno: ino}, nil + } + } + return nil, ErrLockCorrupt +} diff --git a/internal/instanceguard/guard_other.go b/internal/instanceguard/guard_other.go new file mode 100644 index 0000000..f6d6797 --- /dev/null +++ b/internal/instanceguard/guard_other.go @@ -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) error { + return ErrUnsupported +} + +func (l *Lease) currentDB() (uint64, uint64, bool, error) { + return 0, 0, false, fmt.Errorf("%w", ErrUnsupported) +} diff --git a/internal/instanceguard/guard_test.go b/internal/instanceguard/guard_test.go new file mode 100644 index 0000000..1e4a40f --- /dev/null +++ b/internal/instanceguard/guard_test.go @@ -0,0 +1,310 @@ +//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) + } + if err := os.WriteFile(db, []byte("replaced"), 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 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 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 +} diff --git a/internal/instanceguard/guard_unix.go b/internal/instanceguard/guard_unix.go new file mode 100644 index 0000000..32c2649 --- /dev/null +++ b/internal/instanceguard/guard_unix.go @@ -0,0 +1,190 @@ +//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 + } + _, 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, 0o600) + if err != nil { + 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) + } + 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 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\n", dev, ino, instanceID) + 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 +} diff --git a/internal/store/durability.go b/internal/store/durability.go new file mode 100644 index 0000000..b295b3c --- /dev/null +++ b/internal/store/durability.go @@ -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 +} diff --git a/internal/store/durability_test.go b/internal/store/durability_test.go new file mode 100644 index 0000000..5f7beac --- /dev/null +++ b/internal/store/durability_test.go @@ -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) + } +} diff --git a/internal/store/durability_unix_test.go b/internal/store/durability_unix_test.go new file mode 100644 index 0000000..14c4f87 --- /dev/null +++ b/internal/store/durability_unix_test.go @@ -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") + } +} diff --git a/internal/store/instance_guard.go b/internal/store/instance_guard.go new file mode 100644 index 0000000..f5bce17 --- /dev/null +++ b/internal/store/instance_guard.go @@ -0,0 +1,46 @@ +package store + +import ( + "database/sql" + "errors" +) + +// InstanceGuardMarker 是 instance_guard 表的唯一一行。 +type InstanceGuardMarker struct { + DBDev uint64 + DBIno uint64 + LockDev uint64 + LockIno uint64 + InstanceID 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 FROM instance_guard WHERE id=1`). + Scan(&dbDev, &dbIno, &lockDev, &lockIno, &m.InstanceID) + 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) + 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`), + int64(m.DBDev), int64(m.DBIno), int64(m.LockDev), int64(m.LockIno), m.InstanceID) + return err +} diff --git a/internal/store/migrate.go b/internal/store/migrate.go index 6dbac23..7b6f187 100644 --- a/internal/store/migrate.go +++ b/internal/store/migrate.go @@ -778,6 +778,32 @@ func init() { }, "postgres": { `DROP TABLE IF EXISTS agent_system_identity`, }}}, + // U8：活动服务锁的库内标记。锁文件 inode 与本行不一致表示锁文件被重建。 + migration{version: 28, 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 '' + )`, + }, "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 '' + )`, + }}, + down: map[string][]string{"sqlite": { + `DROP TABLE IF EXISTS instance_guard`, + }, "postgres": { + `DROP TABLE IF EXISTS instance_guard`, + }}}, ) } diff --git a/internal/store/store.go b/internal/store/store.go index ba64e01..272542d 100644 --- a/internal/store/store.go +++ b/internal/store/store.go @@ -107,7 +107,8 @@ func openSQLiteFile(path string, createArtifacts bool) (*Store, error) { // 跨进程迁移/写路径依赖 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 @@ -199,6 +200,10 @@ func finishOpen(db *sql.DB, d dialect, artifactsDir string, blobs blobstore.Stor 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()