From 89acd16ca5c46e4a2d2fc58f9e598dccf55be65c Mon Sep 17 00:00:00 2001 From: weichen Date: Mon, 12 Oct 2026 13:53:42 +0000 Subject: [PATCH] Refuse a second SQLite service before migration Why: Two processes opening the same SQLite file both migrate and reconcile, so the second one seals the first one's live runs. The activity lock is taken before open, and durability settings are read back on the application connection. Evidence: SELF go test ./internal/instanceguard ./internal/store ./internal/app (filtered T8 cases) on feature/u8-sqlite-instance-guard. PG durability readback is NOT_RUN locally; Dev CI test-backend-pg is the GATE. Impact: SQLite serve and migrate share one non-blocking flock. A replaced database or rebuilt lock file refuses start unless --adopt-instance-guard. PostgreSQL serve refuses when fsync, full_page_writes, or synchronous_commit is off. Refs: docs/recovery/u8-sqlite-instance-guard.md Co-authored-by: Cursor --- cmd/agent-system/main.go | 9 +- cmd/agent-system/migrate.go | 30 +++ internal/app/app.go | 32 +-- internal/app/guard_test.go | 151 ++++++++++++ internal/app/service_store.go | 80 +++++++ internal/instanceguard/guard.go | 148 ++++++++++++ internal/instanceguard/guard_other.go | 18 ++ internal/instanceguard/guard_test.go | 310 +++++++++++++++++++++++++ internal/instanceguard/guard_unix.go | 190 +++++++++++++++ internal/store/durability.go | 67 ++++++ internal/store/durability_test.go | 95 ++++++++ internal/store/durability_unix_test.go | 35 +++ internal/store/instance_guard.go | 46 ++++ internal/store/migrate.go | 26 +++ internal/store/store.go | 7 +- 15 files changed, 1214 insertions(+), 30 deletions(-) create mode 100644 cmd/agent-system/migrate.go create mode 100644 internal/app/guard_test.go create mode 100644 internal/app/service_store.go create mode 100644 internal/instanceguard/guard.go create mode 100644 internal/instanceguard/guard_other.go create mode 100644 internal/instanceguard/guard_test.go create mode 100644 internal/instanceguard/guard_unix.go create mode 100644 internal/store/durability.go create mode 100644 internal/store/durability_test.go create mode 100644 internal/store/durability_unix_test.go create mode 100644 internal/store/instance_guard.go 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()