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 <cursoragent@cursor.com>
This commit is contained in:
2026-10-12 13:53:42 +00:00
co-authored by Cursor
parent acabb85e2a
commit 89acd16ca5
15 changed files with 1214 additions and 30 deletions
+7 -2
View File
@@ -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,
+30
View File
@@ -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
}
+5 -27
View File
@@ -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 != "" {
+151
View File
@@ -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")
}
+80
View File
@@ -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
}
+148
View File
@@ -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
}
+18
View File
@@ -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)
}
+310
View File
@@ -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
}
+190
View File
@@ -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
}
+67
View File
@@ -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
}
+95
View File
@@ -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)
}
}
+35
View File
@@ -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")
}
}
+46
View File
@@ -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
}
+26
View File
@@ -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`,
}}},
)
}
+6 -1
View File
@@ -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()