112 lines
3.5 KiB
Go
112 lines
3.5 KiB
Go
package app
|
||
|
||
import (
|
||
"errors"
|
||
"fmt"
|
||
"log"
|
||
"os"
|
||
"path/filepath"
|
||
"strings"
|
||
|
||
"agent-system-backend/internal/blobstore"
|
||
"agent-system-backend/internal/instanceguard"
|
||
"agent-system-backend/internal/store"
|
||
)
|
||
|
||
// openServiceStore 打开活动服务使用的存储。
|
||
// SQLite 档先取活动服务锁，再打开（打开即迁移），然后核对守卫标记。
|
||
// 返回的函数先关存储再放锁。
|
||
func openServiceStore(opt Options) (*store.Store, func(), error) {
|
||
nop := func() {}
|
||
if opt.StorageDSN != "" {
|
||
artifacts := opt.ArtifactsDir
|
||
if artifacts == "" {
|
||
artifacts = filepath.Join(opt.DataDir, "artifacts")
|
||
}
|
||
// 共享工件（U5/KTD9）：配置了 S3 兼容对象存储时注入（多实例共享）；
|
||
// 未配置回落本地目录（单实例/共享卷）。凭据只经环境（不入库不入报告）。
|
||
blobs, berr := blobstore.NewS3FromEnv()
|
||
if berr != nil {
|
||
return nil, nop, fmt.Errorf("blobstore s3: %w", berr)
|
||
}
|
||
if blobs != nil {
|
||
log.Printf("artifacts: shared via S3-compatible object storage (prefix %q)", blobs.Prefix)
|
||
}
|
||
st, err := store.OpenPostgresWithOptions(opt.StorageDSN, artifacts, opt.PGPool, blobs)
|
||
if err != nil && opt.Tier == "team" {
|
||
// team 档拒绝就绪（R2）：PG 故障不回落 SQLite，错误必须明说。
|
||
return nil, nop, fmt.Errorf("team 档打开 PostgreSQL 失败，拒绝就绪（不回落 SQLite）: %w", err)
|
||
}
|
||
if err != nil {
|
||
return nil, nop, fmt.Errorf("open store: %w", err)
|
||
}
|
||
if err := verifyConfiguredIdentity(st, "pg"); err != nil {
|
||
_ = st.Close()
|
||
return nil, nop, err
|
||
}
|
||
return st, func() { _ = st.Close() }, nil
|
||
}
|
||
|
||
lease, err := instanceguard.Acquire(opt.DataDir)
|
||
if err != nil {
|
||
return nil, nop, fmt.Errorf("instance guard: %w", err)
|
||
}
|
||
releaseLease := func() { lease.Release() }
|
||
if err := lease.AutomaticRefusal(); err != nil {
|
||
adoptable := opt.AdoptInstanceGuard && (errors.Is(err, instanceguard.ErrDBReplaced) || errors.Is(err, instanceguard.ErrLockCorrupt))
|
||
if !adoptable {
|
||
releaseLease()
|
||
return nil, nop, fmt.Errorf("instance guard: %w", err)
|
||
}
|
||
}
|
||
st, err := store.OpenSQLite(lease.DBPath())
|
||
if err != nil {
|
||
releaseLease()
|
||
return nil, nop, fmt.Errorf("open store: %w", err)
|
||
}
|
||
if err := verifyConfiguredIdentity(st, "sqlite"); err != nil {
|
||
_ = st.Close()
|
||
releaseLease()
|
||
return nil, nop, err
|
||
}
|
||
if err := instanceguard.Bind(st, lease, instanceguard.BindOptions{
|
||
Adopt: opt.AdoptInstanceGuard,
|
||
InstanceID: strings.TrimSpace(os.Getenv("AGENT_SYSTEM_INSTANCE_ID")),
|
||
}); err != nil {
|
||
_ = st.Close()
|
||
releaseLease()
|
||
return nil, nop, fmt.Errorf("instance guard: %w", err)
|
||
}
|
||
return st, func() {
|
||
_ = st.Close()
|
||
lease.Release()
|
||
}, nil
|
||
}
|
||
|
||
// verifyConfiguredIdentity U1 I3：svc 注入的服务实例身份（manifest.instance_id）与库内 marker 对齐；
|
||
// 未设置则跳过（兼容未接线旧调用方）。与 --instance（runs 进程标识）无关。
|
||
func verifyConfiguredIdentity(st *store.Store, defaultRef string) error {
|
||
sysID := strings.TrimSpace(os.Getenv("AGENT_SYSTEM_INSTANCE_ID"))
|
||
if sysID == "" {
|
||
return nil
|
||
}
|
||
dbref := strings.TrimSpace(os.Getenv("AGENT_SYSTEM_DBREF"))
|
||
if dbref == "" {
|
||
dbref = defaultRef
|
||
}
|
||
if err := st.VerifySystemIdentity(sysID, dbref); err != nil {
|
||
return fmt.Errorf("system identity: %w", err)
|
||
}
|
||
return nil
|
||
}
|
||
|
||
// Migrate 执行显式迁移。SQLite 必须先取得活动服务锁，有服务占用则拒绝。
|
||
func Migrate(opt Options) error {
|
||
_, release, err := openServiceStore(opt)
|
||
if err != nil {
|
||
return err
|
||
}
|
||
release()
|
||
return nil
|
||
}
|