diff --git a/cmd/agent-system/main.go b/cmd/agent-system/main.go index 664143f..7df1a50 100644 --- a/cmd/agent-system/main.go +++ b/cmd/agent-system/main.go @@ -12,6 +12,7 @@ import ( "agent-system-backend/internal/app" "agent-system-backend/internal/config" + "agent-system-backend/internal/store" ) var version = "0.1.0-dev" @@ -48,6 +49,12 @@ func main() { execBackend := fs.String("exec-backend", "", "执行后端：direct（默认，现状直启）| sandbox（bwrap 隔离档：断网、只读系统根、仅会话目录可写、进程数限额）") execDepsDir := fs.String("exec-deps-dir", "", "sandbox 档只读依赖层目录（site-packages 等；空=不挂载）") execWorkspace := fs.String("exec-workspace", "", "sandbox 档工作区模式：空=直写会话目录 | overlay=冻结基底+私有 upper+CAS 回收（S0）") + tier := fs.String("tier", "", "部署档：空=auto（DSN→PG，否则 SQLite）| local（SQLite，不允许 DSN）| team（PG fail-closed：缺 DSN/未认证拒绝启动，PG 故障拒绝就绪）") + pgMaxOpen := fs.Int("pg-max-open", 0, "PG 最大打开连接（0=默认 16）") + pgMaxIdle := fs.Int("pg-max-idle", 0, "PG 最大空闲连接（0=默认 8）") + 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=不限；覆盖全部业务查询）") fs.Parse(os.Args[1:]) cliSet := visited(fs) @@ -56,17 +63,31 @@ func main() { StorageDSN: *dsn, ArtifactsDir: *artifacts, AuthMode: *auth, Instance: *instance, InsecureListen: *insecure, AllowedHostsRaw: *allowedHosts, DisableExecTools: *noExec, ExecBackend: *execBackend, ExecDepsDir: *execDepsDir, ExecWorkspace: *execWorkspace, + Tier: *tier, PGMaxOpen: *pgMaxOpen, PGMaxIdle: *pgMaxIdle, + PGConnMaxIdleSec: *pgIdleSec, PGConnMaxLifetimeSec: *pgLifetimeSec, PGQueryTimeoutSec: *pgQueryTimeoutSec, }, cliSet) if err != nil { fmt.Fprintln(os.Stderr, "agent-system:", err) os.Exit(1) } + if msgs := resolved.TierProblems(); len(msgs) > 0 { + for _, msg := range msgs { + fmt.Fprintln(os.Stderr, "agent-system: tier guard:", msg) + } + os.Exit(1) + } if err := app.Run(app.Options{ Addr: resolved.Addr, DataDir: resolved.DataDir, PluginsDir: resolved.PluginsDir, ForceMock: resolved.Mock, 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, + ExecWorkspace: resolved.ExecWorkspace, + Tier: resolved.Tier, + PGPool: store.PoolOptions{ + MaxOpenConn: resolved.PGMaxOpen, MaxIdleConn: resolved.PGMaxIdle, + ConnMaxIdleTimeSec: resolved.PGConnMaxIdleSec, ConnMaxLifetimeSec: resolved.PGConnMaxLifetimeSec, + QueryTimeoutSec: resolved.PGQueryTimeoutSec, + }, }); err != nil { fmt.Fprintln(os.Stderr, "agent-system:", err) os.Exit(1) @@ -85,6 +106,16 @@ func visited(fs *flag.FlagSet) map[string]bool { name = "allowedHosts" case "disable-exec-tools": name = "disableExecTools" + case "pg-max-open": + name = "pgMaxOpen" + case "pg-max-idle": + name = "pgMaxIdle" + case "pg-conn-max-idle-sec": + name = "pgConnMaxIdleSec" + case "pg-conn-max-lifetime-sec": + name = "pgConnMaxLifetimeSec" + case "pg-query-timeout-sec": + name = "pgQueryTimeoutSec" } set[name] = true }) @@ -113,12 +144,20 @@ func runConfig(args []string) int { execBackend := fs.String("exec-backend", "", "") execDepsDir := fs.String("exec-deps-dir", "", "") execWorkspace := fs.String("exec-workspace", "", "") + tier := fs.String("tier", "", "") + pgMaxOpen := fs.Int("pg-max-open", 0, "") + pgMaxIdle := fs.Int("pg-max-idle", 0, "") + pgIdleSec := fs.Int("pg-conn-max-idle-sec", 0, "") + pgLifetimeSec := fs.Int("pg-conn-max-lifetime-sec", 0, "") + pgQueryTimeoutSec := fs.Int("pg-query-timeout-sec", 0, "") fs.Parse(args[1:]) resolved, err := config.Load(*cfgFile, os.LookupEnv, config.Config{ Addr: *addr, DataDir: *data, PluginsDir: *pluginsDir, Mock: *mock, StorageDSN: *dsn, ArtifactsDir: *artifacts, AuthMode: *auth, Instance: *instance, InsecureListen: *insecure, AllowedHostsRaw: *allowedHosts, DisableExecTools: *noExec, ExecBackend: *execBackend, ExecDepsDir: *execDepsDir, ExecWorkspace: *execWorkspace, + Tier: *tier, PGMaxOpen: *pgMaxOpen, PGMaxIdle: *pgMaxIdle, + PGConnMaxIdleSec: *pgIdleSec, PGConnMaxLifetimeSec: *pgLifetimeSec, PGQueryTimeoutSec: *pgQueryTimeoutSec, }, visited(fs)) if err != nil { fmt.Fprintln(os.Stderr, "agent-system:", err) diff --git a/internal/app/app.go b/internal/app/app.go index 227b058..c6fb41a 100644 --- a/internal/app/app.go +++ b/internal/app/app.go @@ -85,6 +85,42 @@ type Options struct { // ExecWorkspace = "overlay" 时 sandbox 档启用 S0 工作区（冻结基底+私有 // upper+CAS 回收）；空 = EXT-16 现状直写会话目录。 ExecWorkspace string + // Tier 部署档（协作存储 KTD1）：空=auto（DSN→PG 否则 SQLite）；local=SQLite； + // team=PG fail-closed（缺 DSN/未认证拒绝启动；PG 故障拒绝就绪，不回落 SQLite）。 + Tier string + // PGPool PG 连接预算（KTD12；零值走默认）。仅 PG 档生效。 + PGPool store.PoolOptions +} + +// tierName 输出部署档名（诊断统一拼写：空 = auto）。 +func tierName(opt Options) string { + if opt.Tier == "" { + return "auto" + } + return opt.Tier +} + +// TierRefusal 部署档门禁（KTD1）：返回阻止启动的问题（空 = 放行）。 +// team 档 fail-closed：缺 DSN、未开认证拒绝；PG 故障的拒绝发生在其后的 +// OpenPostgres（Run 对 team 档的错误显式声明不回落 SQLite）。 +func TierRefusal(opt Options) []string { + return config.Resolved{Config: config.Config{ + Addr: opt.Addr, AuthMode: opt.AuthMode, StorageDSN: opt.StorageDSN, Tier: opt.Tier, + PGMaxOpen: opt.PGPool.MaxOpenConn, PGMaxIdle: opt.PGPool.MaxIdleConn, + PGConnMaxIdleSec: opt.PGPool.ConnMaxIdleTimeSec, PGConnMaxLifetimeSec: opt.PGPool.ConnMaxLifetimeSec, + PGQueryTimeoutSec: opt.PGPool.QueryTimeoutSec, + }}.TierProblems() +} + +// latestVersion 返回已应用迁移的最高版本（0 = 空账本）。 +func latestVersion(versions []int) int { + max := 0 + for _, v := range versions { + if v > max { + max = v + } + } + return max } // Run 装配并阻塞服务直至 SIGINT/SIGTERM（优雅停机：中断在途 run、冲净事件缓冲）。 @@ -93,13 +129,25 @@ func Run(opt Options) error { Service: "agent-system", Module: "app", Level: logsys.LevelInfo, Sinks: []logsys.Sink{&sink.Console{Out: os.Stderr, Format: "json"}}, }) - logger.Info("agent-system starting", "addr", opt.Addr, "data", opt.DataDir) + logger.Info("agent-system starting", "addr", opt.Addr, "data", opt.DataDir, "tier", tierName(opt)) resolved := config.Resolved{Config: config.Config{ Addr: opt.Addr, AuthMode: opt.AuthMode, StorageDSN: opt.StorageDSN, InsecureListen: opt.InsecureListen, + Tier: opt.Tier, PGMaxOpen: opt.PGPool.MaxOpenConn, PGMaxIdle: opt.PGPool.MaxIdleConn, + PGConnMaxIdleSec: opt.PGPool.ConnMaxIdleTimeSec, PGConnMaxLifetimeSec: opt.PGPool.ConnMaxLifetimeSec, + PGQueryTimeoutSec: opt.PGPool.QueryTimeoutSec, }} if msg := resolved.ListenRefusal(); msg != "" { return fmt.Errorf("listen guard: %s", msg) } + // 部署档门禁（KTD1 fail-closed）：team 档缺 DSN/未开认证直接拒绝启动； + // local 档带 DSN 是配置错误。此处与 config.TierProblems 同规则（纵深防御， + // 服务入口不依赖 CLI 是否校验）。 + if msgs := TierRefusal(opt); len(msgs) > 0 { + return fmt.Errorf("tier guard: %s", strings.Join(msgs, "；")) + } + if opt.Tier == "team" && !opt.DisableExecTools && opt.ExecBackend == "" { + log.Printf("warning: team 档未配置执行隔离（disableExecTools 或 execBackend）；向真实多用户开放前必须补齐（方案 KTD1）") + } if opt.InsecureListen && !config.IsLoopbackListen(opt.Addr) && !resolved.AuthEnabled() { log.Printf("warning: listening on %s without authentication because insecureListen is set", opt.Addr) } @@ -110,7 +158,11 @@ func Run(opt Options) error { if artifacts == "" { artifacts = filepath.Join(opt.DataDir, "artifacts") } - st, err = store.OpenPostgres(opt.StorageDSN, artifacts) + st, err = store.OpenPostgresWithOptions(opt.StorageDSN, artifacts, opt.PGPool, nil) + 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")) } @@ -118,7 +170,15 @@ func Run(opt Options) error { return fmt.Errorf("open store: %w", err) } defer st.Close() - log.Printf("storage tier: %s", st.Dialect()) + pool := st.Pool() + if versions, verr := st.MigrationStatus(); verr == nil { + log.Printf("storage tier: %s (dialect=%s pool=%d/%d idleSec=%d lifetimeSec=%d schema=v%d)", + tierName(opt), st.Dialect(), pool.MaxOpenConn, pool.MaxIdleConn, pool.ConnMaxIdleTimeSec, + pool.ConnMaxLifetimeSec, latestVersion(versions)) + } else { + log.Printf("storage tier: %s (dialect=%s pool=%d/%d schema=? migration status unavailable: %v)", + tierName(opt), st.Dialect(), pool.MaxOpenConn, pool.MaxIdleConn, verr) + } b := bus.New() reg := tools.NewRegistry() diff --git a/internal/app/tier_test.go b/internal/app/tier_test.go new file mode 100644 index 0000000..690fc72 --- /dev/null +++ b/internal/app/tier_test.go @@ -0,0 +1,31 @@ +package app + +import ( + "strings" + "testing" +) + +// 协作存储方案 U2（KTD1/R2）：服务入口的部署档门禁。 +// team 档缺 DSN / 未开认证必须拒绝启动——不创建 SQLite、不进入装配。 +func TestTierRefusalTeam(t *testing.T) { + msgs := TierRefusal(Options{Tier: "team", Addr: "127.0.0.1:8788"}) + if len(msgs) != 2 { + t.Fatalf("team without dsn/auth: %v", msgs) + } + joined := strings.Join(msgs, "；") + if !strings.Contains(joined, "PostgreSQL DSN") || !strings.Contains(joined, "认证") { + t.Fatalf("refusal wording: %v", msgs) + } + msgs = TierRefusal(Options{Tier: "team", Addr: "127.0.0.1:8788", StorageDSN: "postgres://u:p@h/db", AuthMode: "on"}) + if len(msgs) != 0 { + t.Fatalf("team with dsn+auth should pass: %v", msgs) + } + msgs = TierRefusal(Options{Tier: "local", Addr: "127.0.0.1:8788", StorageDSN: "postgres://u:p@h/db"}) + if len(msgs) != 1 || !strings.Contains(msgs[0], "local 档不使用") { + t.Fatalf("local with dsn: %v", msgs) + } + msgs = TierRefusal(Options{Addr: "127.0.0.1:8788"}) + if len(msgs) != 0 { + t.Fatalf("auto default: %v", msgs) + } +} diff --git a/internal/config/config.go b/internal/config/config.go index 0e8a146..45defa6 100644 --- a/internal/config/config.go +++ b/internal/config/config.go @@ -11,6 +11,7 @@ import ( "io" "net" "os" + "strconv" "strings" ) @@ -35,6 +36,17 @@ type Config struct { ExecDepsDir string `json:"execDepsDir"` // ExecWorkspace sandbox 档工作区模式：空=直写；overlay=冻结基底+upper。 ExecWorkspace string `json:"execWorkspace"` + // Tier 部署档（协作存储方案 KTD1）：空=auto（DSN→PG，否则 SQLite）； + // local=强制 SQLite（不允许 DSN）；team=强制 PG（缺 DSN/未开认证拒绝启动， + // PG 故障拒绝就绪，绝不回落 SQLite）。 + Tier string `json:"tier"` + // PG 连接预算（KTD12）：零值走默认（16/8/300s/不限/不限）；负值非法。 + // PGQueryTimeoutSec>0 时以 statement_timeout 附着到 PG DSN。 + PGMaxOpen int `json:"pgMaxOpen"` + PGMaxIdle int `json:"pgMaxIdle"` + PGConnMaxIdleSec int `json:"pgConnMaxIdleSec"` + PGConnMaxLifetimeSec int `json:"pgConnMaxLifetimeSec"` + PGQueryTimeoutSec int `json:"pgQueryTimeoutSec"` } var defaults = Config{ @@ -46,20 +58,26 @@ var defaults = Config{ // envKey 字段 → 环境变量名（AGENT_SYSTEM_ 前缀）。 var envKey = map[string]string{ - "addr": "AGENT_SYSTEM_ADDR", - "data": "AGENT_SYSTEM_DATA", - "plugins": "AGENT_SYSTEM_PLUGINS", - "dsn": "AGENT_SYSTEM_DSN", - "artifacts": "AGENT_SYSTEM_ARTIFACTS", - "auth": "AGENT_SYSTEM_AUTH", - "mock": "AGENT_SYSTEM_MOCK", - "instance": "AGENT_SYSTEM_INSTANCE", - "execBackend": "AGENT_SYSTEM_EXEC_BACKEND", - "execDepsDir": "AGENT_SYSTEM_EXEC_DEPS_DIR", - "execWorkspace": "AGENT_SYSTEM_EXEC_WORKSPACE", - "insecureListen": "AGENT_SYSTEM_INSECURE_LISTEN", - "allowedHosts": "AGENT_SYSTEM_ALLOWED_HOSTS", - "disableExecTools": "AGENT_SYSTEM_DISABLE_EXEC_TOOLS", + "addr": "AGENT_SYSTEM_ADDR", + "data": "AGENT_SYSTEM_DATA", + "plugins": "AGENT_SYSTEM_PLUGINS", + "dsn": "AGENT_SYSTEM_DSN", + "artifacts": "AGENT_SYSTEM_ARTIFACTS", + "auth": "AGENT_SYSTEM_AUTH", + "mock": "AGENT_SYSTEM_MOCK", + "instance": "AGENT_SYSTEM_INSTANCE", + "execBackend": "AGENT_SYSTEM_EXEC_BACKEND", + "execDepsDir": "AGENT_SYSTEM_EXEC_DEPS_DIR", + "execWorkspace": "AGENT_SYSTEM_EXEC_WORKSPACE", + "insecureListen": "AGENT_SYSTEM_INSECURE_LISTEN", + "allowedHosts": "AGENT_SYSTEM_ALLOWED_HOSTS", + "disableExecTools": "AGENT_SYSTEM_DISABLE_EXEC_TOOLS", + "tier": "AGENT_SYSTEM_TIER", + "pgMaxOpen": "AGENT_SYSTEM_PG_MAX_OPEN", + "pgMaxIdle": "AGENT_SYSTEM_PG_MAX_IDLE", + "pgConnMaxIdleSec": "AGENT_SYSTEM_PG_CONN_MAX_IDLE_SEC", + "pgConnMaxLifetimeSec": "AGENT_SYSTEM_PG_CONN_MAX_LIFETIME_SEC", + "pgQueryTimeoutSec": "AGENT_SYSTEM_PG_QUERY_TIMEOUT_SEC", } // Source 配置来源层级（数值越大优先级越高）。 @@ -109,11 +127,7 @@ func Load(file string, envLookup func(string) (string, bool), cli Config, cliSet } // 文件层按"键存在"覆盖（区分未出现与显式 false/空串；空值有效性由字段语义判定）。 for f := range present { - if isBoolField(f) { - setBool(&cfg, f, boolField(fc, f)) - } else { - setString(&cfg, f, stringField(fc, f)) - } + setField(&cfg, f, fc) src[f] = SrcFile } } @@ -126,6 +140,12 @@ func Load(file string, envLookup func(string) (string, bool), cli Config, cliSet return Resolved{}, err } setBool(&cfg, f, b) + } else if isIntField(f) { + n, err := parseInt(key, v) + if err != nil { + return Resolved{}, err + } + setInt(&cfg, f, n) } else { setString(&cfg, f, v) } @@ -135,15 +155,22 @@ func Load(file string, envLookup func(string) (string, bool), cli Config, cliSet // CLI 层：仅显式传入。 for f := range cliSet { src[f] = SrcCLI - if isBoolField(f) { - setBool(&cfg, f, boolField(cli, f)) - } else { - setString(&cfg, f, stringField(cli, f)) - } + setField(&cfg, f, cli) } return Resolved{Config: cfg, sources: src}, nil } +// setField 从同类型快照覆盖单个字段（文件/CLI 层复用；字段类型由字段名决定）。 +func setField(dst *Config, f string, src Config) { + if isBoolField(f) { + setBool(dst, f, boolField(src, f)) + } else if isIntField(f) { + setInt(dst, f, intField(src, f)) + } else { + setString(dst, f, stringField(src, f)) + } +} + func loadFile(path string) (Config, map[string]bool, error) { raw, err := os.ReadFile(path) if err != nil { @@ -186,6 +213,18 @@ func loadFile(path string) (Config, map[string]bool, error) { dst = &c.ExecDepsDir case "execWorkspace": dst = &c.ExecWorkspace + case "tier": + dst = &c.Tier + case "pgMaxOpen": + dst = &c.PGMaxOpen + case "pgMaxIdle": + dst = &c.PGMaxIdle + case "pgConnMaxIdleSec": + dst = &c.PGConnMaxIdleSec + case "pgConnMaxLifetimeSec": + dst = &c.PGConnMaxLifetimeSec + case "pgQueryTimeoutSec": + dst = &c.PGQueryTimeoutSec default: return Config{}, nil, fmt.Errorf("config file %s: unknown key %q", path, k) } @@ -221,6 +260,8 @@ func stringField(c Config, f string) string { return c.ExecDepsDir case "execWorkspace": return c.ExecWorkspace + case "tier": + return c.Tier } return "" } @@ -229,6 +270,45 @@ func isBoolField(f string) bool { return f == "mock" || f == "insecureListen" || f == "disableExecTools" } +func isIntField(f string) bool { + switch f { + case "pgMaxOpen", "pgMaxIdle", "pgConnMaxIdleSec", "pgConnMaxLifetimeSec", "pgQueryTimeoutSec": + return true + } + return false +} + +func intField(c Config, f string) int { + switch f { + case "pgMaxOpen": + return c.PGMaxOpen + case "pgMaxIdle": + return c.PGMaxIdle + case "pgConnMaxIdleSec": + return c.PGConnMaxIdleSec + case "pgConnMaxLifetimeSec": + return c.PGConnMaxLifetimeSec + case "pgQueryTimeoutSec": + return c.PGQueryTimeoutSec + } + return 0 +} + +func setInt(c *Config, f string, v int) { + switch f { + case "pgMaxOpen": + c.PGMaxOpen = v + case "pgMaxIdle": + c.PGMaxIdle = v + case "pgConnMaxIdleSec": + c.PGConnMaxIdleSec = v + case "pgConnMaxLifetimeSec": + c.PGConnMaxLifetimeSec = v + case "pgQueryTimeoutSec": + c.PGQueryTimeoutSec = v + } +} + func boolField(c Config, f string) bool { switch f { case "insecureListen": @@ -275,6 +355,8 @@ func setString(c *Config, f, v string) { c.ExecDepsDir = v case "execWorkspace": c.ExecWorkspace = v + case "tier": + c.Tier = v } } @@ -288,6 +370,14 @@ func parseBool(name, v string) (bool, error) { return false, fmt.Errorf("env %s: invalid bool %q", name, v) } +func parseInt(name, v string) (int, error) { + n, err := strconv.Atoi(strings.TrimSpace(v)) + if err != nil { + return 0, fmt.Errorf("env %s: invalid int %q", name, v) + } + return n, nil +} + // Explain 只读诊断：逐字段打印解析值与来源层级，DSN 脱敏。 func (r Resolved) Explain(w io.Writer, file string) { if file != "" { @@ -296,16 +386,32 @@ func (r Resolved) Explain(w io.Writer, file string) { fmt.Fprintln(w, "config file: (none)") } fmt.Fprintln(w, "field value source") - for _, f := range []string{"addr", "data", "plugins", "dsn", "artifacts", "auth", "mock", "instance", "insecureListen", "allowedHosts", "disableExecTools", "execBackend"} { - v := stringField(r.Config, f) + for _, f := range []string{"addr", "data", "plugins", "dsn", "artifacts", "auth", "mock", "instance", "insecureListen", "allowedHosts", "disableExecTools", "execBackend", "tier", "pgMaxOpen", "pgMaxIdle", "pgConnMaxIdleSec", "pgConnMaxLifetimeSec", "pgQueryTimeoutSec"} { + var v string if isBoolField(f) { v = fmt.Sprintf("%t", boolField(r.Config, f)) + } else if isIntField(f) { + n := intField(r.Config, f) + if d := intDefault(f); n == 0 && d != 0 { + v = fmt.Sprintf("%d (默认)", d) + } else { + v = fmt.Sprintf("%d", n) + } + } else { + v = stringField(r.Config, f) } if f == "dsn" { v = redact(v) } fmt.Fprintf(w, "%-16s %-56s %s\n", f, v, r.sources[f]) } + if msgs := r.TierProblems(); len(msgs) > 0 { + for _, msg := range msgs { + fmt.Fprintf(w, "tier guard: 拒绝启动（%s）\n", msg) + } + } else { + fmt.Fprintln(w, "tier guard: 放行") + } if msg := r.ListenRefusal(); msg != "" { fmt.Fprintf(w, "listen guard: 拒绝启动（%s）\n", msg) } else if r.InsecureListen && !IsLoopbackListen(r.Addr) && !r.AuthEnabled() { @@ -320,6 +426,90 @@ func (r Resolved) AuthEnabled() bool { return r.AuthMode == "on" || (r.AuthMode != "off" && r.StorageDSN != "") } +// intDefault 返回 int 字段的默认值（0 表示零值合法、无默认替换）。 +func intDefault(f string) int { + switch f { + case "pgMaxOpen": + return DefaultPGMaxOpen + case "pgMaxIdle": + return DefaultPGMaxIdle + case "pgConnMaxIdleSec": + return DefaultPGConnMaxIdleSec + } + return 0 +} + +// 连接预算默认值（KTD12；与 store 层默认保持一致，两处同时修改时以测试对齐）。 +const ( + DefaultPGMaxOpen = 16 + DefaultPGMaxIdle = 8 + DefaultPGConnMaxIdleSec = 300 +) + +// TierProblems 部署档语义校验：返回阻止启动的问题列表（空 = 放行）。 +// team 档 fail-closed：缺 DSN、未开认证直接拒绝；local 档不允许 DSN（KTD1）。 +// 执行隔离在 team 档是启动警告而非拒绝（本增量先可见、U3/U5 再硬化）。 +func (r Resolved) TierProblems() []string { + var msgs []string + switch r.Tier { + case "", "local", "team": + default: + msgs = append(msgs, fmt.Sprintf("未知部署档 %q（local | team）", r.Tier)) + } + if r.Tier == "team" { + if r.StorageDSN == "" { + msgs = append(msgs, "team 档需要 PostgreSQL DSN（dsn / AGENT_SYSTEM_DSN）；不会回落 SQLite") + } + if !r.AuthEnabled() { + msgs = append(msgs, "team 档必须开启认证（auth=on）") + } + } + if r.Tier == "local" && r.StorageDSN != "" { + msgs = append(msgs, "local 档不使用 PostgreSQL DSN（改用 team 档，或去掉 dsn）") + } + if v := r.PGMaxOpen; v < 0 { + msgs = append(msgs, fmt.Sprintf("pgMaxOpen 不能为负（当前 %d）", v)) + } + if v := r.PGMaxIdle; v < 0 { + msgs = append(msgs, fmt.Sprintf("pgMaxIdle 不能为负（当前 %d）", v)) + } + if v := r.PGConnMaxIdleSec; v < 0 { + msgs = append(msgs, fmt.Sprintf("pgConnMaxIdleSec 不能为负（当前 %d）", v)) + } + if v := r.PGConnMaxLifetimeSec; v < 0 { + msgs = append(msgs, fmt.Sprintf("pgConnMaxLifetimeSec 不能为负（当前 %d）", v)) + } + if v := r.PGQueryTimeoutSec; v < 0 { + msgs = append(msgs, fmt.Sprintf("pgQueryTimeoutSec 不能为负（当前 %d）", v)) + } + if idle, open := r.EffectivePGMaxIdle(), r.EffectivePGMaxOpen(); idle > open { + msgs = append(msgs, fmt.Sprintf("pgMaxIdle(%d) 大于 pgMaxOpen(%d)", idle, open)) + } + return msgs +} + +// EffectivePGMaxOpen/Idle/ConnMaxIdleSec 返回连接预算有效值（零值落默认，KTD12）。 +func (r Resolved) EffectivePGMaxOpen() int { + if r.PGMaxOpen > 0 { + return r.PGMaxOpen + } + return DefaultPGMaxOpen +} + +func (r Resolved) EffectivePGMaxIdle() int { + if r.PGMaxIdle > 0 { + return r.PGMaxIdle + } + return DefaultPGMaxIdle +} + +func (r Resolved) EffectivePGConnMaxIdleSec() int { + if r.PGConnMaxIdleSec > 0 { + return r.PGConnMaxIdleSec + } + return DefaultPGConnMaxIdleSec +} + // AllowedHosts 把逗号分隔的主机名拆开，忽略空段。 func (r Resolved) AllowedHosts() []string { var out []string diff --git a/internal/config/config_test.go b/internal/config/config_test.go index 1ddfaf1..08a3cd5 100644 --- a/internal/config/config_test.go +++ b/internal/config/config_test.go @@ -4,6 +4,7 @@ import ( "bytes" "os" "path/filepath" + "strings" "testing" ) @@ -220,3 +221,135 @@ func TestListenGuard(t *testing.T) { t.Fatalf("auth on should pass: %+v %q", authed.Config, authed.ListenRefusal()) } } + +// ---- 协作存储方案 U2：部署档（KTD1）与连接预算（KTD12）---- + +// team 档 fail-closed：缺 DSN、未开认证都拒绝；DSN+auth=on 放行。 +func TestTierTeamFailClosed(t *testing.T) { + r, err := Load("", noEnv, Config{Tier: "team"}, map[string]bool{"tier": true}) + if err != nil { + t.Fatal(err) + } + msgs := r.TierProblems() + if len(msgs) != 2 { + t.Fatalf("team without dsn/auth should refuse with 2 problems, got %v", msgs) + } + joined := msgs[0] + msgs[1] + if !strings.Contains(joined, "PostgreSQL DSN") || !strings.Contains(joined, "认证") { + t.Fatalf("refusal messages wrong: %v", msgs) + } + + // 补上 DSN（auto 认证随 DSN 开启）→ 放行。 + r, err = Load("", noEnv, Config{Tier: "team", StorageDSN: "postgres://u:p@h/db"}, map[string]bool{"tier": true, "dsn": true}) + if err != nil { + t.Fatal(err) + } + if msgs = r.TierProblems(); len(msgs) != 0 { + t.Fatalf("team with dsn should pass, got %v", msgs) + } + + // 显式关认证仍然拒绝。 + r, err = Load("", noEnv, Config{Tier: "team", StorageDSN: "postgres://u:p@h/db", AuthMode: "off"}, map[string]bool{"tier": true, "dsn": true, "auth": true}) + if err != nil { + t.Fatal(err) + } + if msgs = r.TierProblems(); len(msgs) != 1 || !strings.Contains(msgs[0], "认证") { + t.Fatalf("team with auth=off should refuse, got %v", msgs) + } +} + +// local 档不允许 DSN；未知档名拒绝。 +func TestTierLocalAndUnknown(t *testing.T) { + r, err := Load("", noEnv, Config{Tier: "local", StorageDSN: "postgres://u:p@h/db"}, map[string]bool{"tier": true, "dsn": true}) + if err != nil { + t.Fatal(err) + } + if msgs := r.TierProblems(); len(msgs) != 1 || !strings.Contains(msgs[0], "local 档不使用") { + t.Fatalf("local with dsn should refuse, got %v", msgs) + } + r, err = Load("", noEnv, Config{Tier: "cluster"}, map[string]bool{"tier": true}) + if err != nil { + t.Fatal(err) + } + if msgs := r.TierProblems(); len(msgs) != 1 || !strings.Contains(msgs[0], "未知部署档") { + t.Fatalf("unknown tier should refuse, got %v", msgs) + } + // auto（空）档维持现状语义：无 DSN 不拒绝。 + r, err = Load("", noEnv, Config{}, nil) + if err != nil { + t.Fatal(err) + } + if msgs := r.TierProblems(); len(msgs) != 0 { + t.Fatalf("auto default should pass, got %v", msgs) + } +} + +// 连接预算四源解析与环境变量整数校验；MaxIdle>MaxOpen 拒绝。 +func TestPGPoolConfigLayers(t *testing.T) { + env := func(k string) (string, bool) { + switch k { + case "AGENT_SYSTEM_PG_MAX_OPEN": + return "7", true + case "AGENT_SYSTEM_PG_MAX_IDLE": + return "9", true + } + return "", false + } + f := fileWith(t, `{"pgMaxOpen":5}`) + r, err := Load(f, env, Config{PGMaxOpen: 3, PGMaxIdle: 2}, map[string]bool{"pgMaxIdle": true}) + if err != nil { + t.Fatal(err) + } + // 文件 5 被 env 7 覆盖（CLI 值 3 未显式传入不参与）；pgMaxIdle CLI 显式 2 胜。 + if r.PGMaxOpen != 7 || r.SourceOf("pgMaxOpen") != SrcEnv { + t.Fatalf("pgMaxOpen = %d (%v), want env 7", r.PGMaxOpen, r.SourceOf("pgMaxOpen")) + } + if r.PGMaxIdle != 2 || r.SourceOf("pgMaxIdle") != SrcCLI { + t.Fatalf("pgMaxIdle = %d (%v), want CLI 2", r.PGMaxIdle, r.SourceOf("pgMaxIdle")) + } + if msgs := r.TierProblems(); len(msgs) != 0 { + t.Fatalf("2/2 pool valid, got %v", msgs) + } + // 有效值：零值落默认。 + r, err = Load("", noEnv, Config{}, nil) + if err != nil { + t.Fatal(err) + } + if r.EffectivePGMaxOpen() != DefaultPGMaxOpen || r.EffectivePGMaxIdle() != DefaultPGMaxIdle || r.EffectivePGConnMaxIdleSec() != DefaultPGConnMaxIdleSec { + t.Fatalf("effective defaults wrong: %+v", r.Config) + } + // idle > open 拒绝。 + r, err = Load("", noEnv, Config{PGMaxOpen: 2, PGMaxIdle: 3}, map[string]bool{"pgMaxOpen": true, "pgMaxIdle": true}) + if err != nil { + t.Fatal(err) + } + if msgs := r.TierProblems(); len(msgs) != 1 || !strings.Contains(msgs[0], "pgMaxIdle(3) 大于 pgMaxOpen(2)") { + t.Fatalf("idle>open should refuse, got %v", msgs) + } + // 非法整数环境变量报错定位到变量。 + bad := func(k string) (string, bool) { + if k == "AGENT_SYSTEM_PG_MAX_OPEN" { + return "many", true + } + return "", false + } + if _, err := Load("", bad, Config{}, nil); err == nil || !strings.Contains(err.Error(), "AGENT_SYSTEM_PG_MAX_OPEN") { + t.Fatalf("invalid int env should fail with var name, got %v", err) + } +} + +// explain 输出包含部署档与预算行，且 tier guard 结论可见（脱敏诊断）。 +func TestExplainTierAndPool(t *testing.T) { + r, err := Load("", noEnv, Config{Tier: "team"}, map[string]bool{"tier": true}) + if err != nil { + t.Fatal(err) + } + var b bytes.Buffer + r.Explain(&b, "") + out := b.String() + for _, want := range []string{"tier", "pgMaxOpen", "pgMaxIdle", "pgConnMaxIdleSec", "pgConnMaxLifetimeSec", "pgQueryTimeoutSec", "tier guard: 拒绝启动"} { + if !strings.Contains(out, want) { + t.Fatalf("explain missing %q:\n%s", want, out) + } + } +} diff --git a/internal/pgbus/pgbus_test.go b/internal/pgbus/pgbus_test.go index 9956d03..920c954 100644 --- a/internal/pgbus/pgbus_test.go +++ b/internal/pgbus/pgbus_test.go @@ -29,6 +29,12 @@ func pgTestStore(t *testing.T) (*store.Store, string) { if err != nil { t.Fatal(err) } + // pg_trgm 预装 public（幂等），理由见 store 包 openTestStore 注释： + // 避免扩展落进用例 schema 后随 CASCADE 删除造成的并行竞态。 + if _, err := admin.Exec(`CREATE EXTENSION IF NOT EXISTS pg_trgm`); err != nil { + admin.Close() + t.Fatalf("preinstall pg_trgm: %v", err) + } if _, err := admin.Exec("CREATE SCHEMA " + schema); err != nil { admin.Close() t.Fatal(err) @@ -37,7 +43,7 @@ func pgTestStore(t *testing.T) (*store.Store, string) { admin.Exec("DROP SCHEMA IF EXISTS " + schema + " CASCADE") admin.Close() }) - full := dsn + "&search_path=" + schema + full := store.AppendPGRuntimeParam(dsn, "search_path", schema+",public") st, err := store.OpenPostgres(full, filepath.Join(t.TempDir(), "artifacts")) if err != nil { t.Fatal(err) diff --git a/internal/runmgr/recovery_test.go b/internal/runmgr/recovery_test.go index d976933..7712254 100644 --- a/internal/runmgr/recovery_test.go +++ b/internal/runmgr/recovery_test.go @@ -34,6 +34,12 @@ func openRecoveryStore(t *testing.T) *store.Store { if err != nil { t.Fatal(err) } + // pg_trgm 预装 public（幂等）：避免 v4 扩展落进用例 schema 后随 CASCADE + // 删除造成的并行竞态（见 store 包 openTestStore 同名注释）。 + if _, err := admin.Exec(`CREATE EXTENSION IF NOT EXISTS pg_trgm`); err != nil { + admin.Close() + t.Fatalf("preinstall pg_trgm: %v", err) + } if _, err := admin.Exec("CREATE SCHEMA " + schema); err != nil { admin.Close() t.Fatalf("create schema: %v", err) @@ -42,7 +48,10 @@ func openRecoveryStore(t *testing.T) *store.Store { admin.Exec("DROP SCHEMA IF EXISTS " + schema + " CASCADE") admin.Close() }) - st, err := store.OpenPostgres(dsn+"&search_path="+schema, filepath.Join(t.TempDir(), "artifacts")) + // 共用 store 的 DSN 运行参数拼接（修复盲拼 "&search_path="：对无查询串的 + // URL DSN 会把参数并进库名，连不上库——svc.sh 形态 DSN 的既有缺陷）。 + schemaDSN := store.AppendPGRuntimeParam(dsn, "search_path", schema+",public") + st, err := store.OpenPostgres(schemaDSN, filepath.Join(t.TempDir(), "artifacts")) if err != nil { t.Fatal(err) } diff --git a/internal/store/messages.go b/internal/store/messages.go index 205f464..a34cc26 100644 --- a/internal/store/messages.go +++ b/internal/store/messages.go @@ -40,7 +40,7 @@ func (s *Store) AppendMessageWithRequest(m *core.Message, requestID string) erro m.CreatedAt = m.CreatedAt.UTC() insert := `INSERT INTO messages(id,conversation_id,run_id,role,parts,seq,created_at,request_id) SELECT ?,?,?,?,?,COALESCE(MAX(seq),0)+1,?,? FROM messages WHERE conversation_id=?` - for attempt := 0; attempt < 8; attempt++ { + for attempt := 0; attempt < 32; attempt++ { _, err := s.db.Exec(s.q(insert), m.ID, m.ConversationID, m.RunID, string(m.Role), marshalJSON(m.Parts), nowISO(), requestID, m.ConversationID) if err == nil { s.TouchConversation(m.ConversationID) @@ -56,8 +56,14 @@ func (s *Store) AppendMessageWithRequest(m *core.Message, requestID string) erro return &DuplicateRequestError{Existing: existing} } } - // 并发同会话追加冲突（PG 多连接竞态）：短抖动退避后重算序号。 - time.Sleep(time.Duration(attempt) * time.Millisecond) + // 并发同会话追加冲突（PG 多连接竞态）：平方退避（封顶 25ms）后重算 + // 序号。真库 24 路并发实测：8 次 × 0–7ms 线性退避会耗尽重试（真实 + // PG 门禁暴露的既有缺陷）；事务内会话锁的根治收口在 U3（KTD3/KTD7）。 + backoff := attempt * attempt + if backoff > 25 { + backoff = 25 + } + time.Sleep(time.Duration(backoff) * time.Millisecond) } return fmt.Errorf("store: message seq conflict retries exhausted (conversation %s)", m.ConversationID) } diff --git a/internal/store/migrate.go b/internal/store/migrate.go index 89eba48..180cb76 100644 --- a/internal/store/migrate.go +++ b/internal/store/migrate.go @@ -1,6 +1,7 @@ package store import ( + "context" "fmt" ) @@ -520,11 +521,51 @@ var v16Down = []string{ } func (s *Store) migrate() error { - if _, err := s.db.Exec(`CREATE TABLE IF NOT EXISTS schema_migrations (version INTEGER PRIMARY KEY, name TEXT NOT NULL, applied_at TEXT NOT NULL)`); err != nil { + if s.dialect.name() == "postgres" { + return s.migratePG() + } + // SQLite：_txlock=immediate（OpenSQLite DSN）保证迁移事务在 BEGIN 即取写锁， + // 跨进程迁移者按 busy_timeout 排队；后续版本读取发生在同一事务内， + // 前一个迁移者提交后本进程看到的已是完整账本（无部分版本可见）。 + return s.migratePass() +} + +// migratePG 以会话级咨询锁串行化并发迁移者（KTD8）：锁在专用连接上获取， +// 等锁者在轮到自己后重新读取账本——前一个迁移者已完成时本轮为空操作。 +// 锁与迁移事务分离：迁移语句走连接池，锁只负责让 migrator 全局互斥。 +func (s *Store) migratePG() error { + ctx := context.Background() + conn, err := s.db.Conn(ctx) + if err != nil { + return fmt.Errorf("migrate lock: %w", err) + } + defer conn.Close() + const lockKey = "'agent-system:schema-migrate'" + if _, err := conn.ExecContext(ctx, `SELECT pg_advisory_lock(hashtext(`+lockKey+`))`); err != nil { + return fmt.Errorf("migrate lock: %w", err) + } + defer conn.ExecContext(ctx, `SELECT pg_advisory_unlock(hashtext(`+lockKey+`))`) + return s.migratePass() +} + +// migratePass 单事务执行全部待应用迁移：DDL 与版本登记同事务提交（KTD8）， +// 中途失败整体回滚——不存在"DDL 已生效而账本未记"的中间态；两步间崩溃 +// 的历史缺口（先提交 DDL 再单独 INSERT 版本）由此消除。 +func (s *Store) migratePass() error { + tx, err := s.db.Begin() + if err != nil { + return fmt.Errorf("migrate: %w", err) + } + defer tx.Rollback() + if _, err := tx.Exec(`CREATE TABLE IF NOT EXISTS schema_migrations (version INTEGER PRIMARY KEY, name TEXT NOT NULL, applied_at TEXT NOT NULL)`); err != nil { return fmt.Errorf("migrate: %w", err) } applied := map[int]bool{} - rows, err := s.db.Query(`SELECT version FROM schema_migrations`) + known := map[int]bool{} + for _, m := range migrations { + known[m.version] = true + } + rows, err := tx.Query(`SELECT version FROM schema_migrations`) if err != nil { return fmt.Errorf("migrate: %w", err) } @@ -534,6 +575,12 @@ func (s *Store) migrate() error { rows.Close() return fmt.Errorf("migrate: %w", err) } + // 兼容门禁（KTD8 应用侧只核对兼容版本）：库里有本二进制不认识的版本 + // 说明库由更新的构建迁移过——拒绝启动，不冒险在未知 schema 上服务。 + if !known[v] { + rows.Close() + return fmt.Errorf("migrate: 数据库 schema 版本 v%d 超出本构建（最高 v%d）；拒绝启动，请升级二进制", v, latestMigrationVersion()) + } applied[v] = true } if err := rows.Err(); err != nil { @@ -549,14 +596,28 @@ func (s *Store) migrate() error { if len(stmts) == 0 { return fmt.Errorf("migrate: v%d %s 缺少 %s 方言", m.version, m.name, s.dialect.name()) } - if err := s.execInTx(stmts); err != nil { - return fmt.Errorf("migrate v%d (%s): %w", m.version, m.name, err) + for _, q := range stmts { + if _, err := tx.Exec(s.q(q)); err != nil { + return fmt.Errorf("migrate v%d (%s): %w", m.version, m.name, err) + } } - if _, err := s.db.Exec(s.q(`INSERT INTO schema_migrations(version,name,applied_at) VALUES(?,?,?)`), m.version, m.name, nowISO()); err != nil { - return fmt.Errorf("migrate: %w", err) + // 故障注入点：DDL 已执行、版本未登记——历史实现在此处崩溃会留下 + // 「DDL 生效而账本缺失」的中间态；同事务设计下这里失败必须整体回滚。 + if fault := s.takeMigrateFault(); fault != nil { + return fmt.Errorf("migrate v%d (%s) 注入故障: %w", m.version, m.name, fault) + } + if _, err := tx.Exec(s.q(`INSERT INTO schema_migrations(version,name,applied_at) VALUES(?,?,?)`), m.version, m.name, nowISO()); err != nil { + return fmt.Errorf("migrate v%d (%s) 版本登记: %w", m.version, m.name, err) } } - return nil + return tx.Commit() +} + +func latestMigrationVersion() int { + if len(migrations) == 0 { + return 0 + } + return migrations[len(migrations)-1].version } func (s *Store) execInTx(stmts []string) error { @@ -592,6 +653,7 @@ func (s *Store) MigrationStatus() ([]int, error) { } // MigrateDown 回滚到指定版本（不含；运维入口，多租户数据折回限制见各 down 注释）。 +// 回滚语句与版本删除同事务（KTD8 原子性对称）。 func (s *Store) MigrateDown(toVersion int) error { applied, err := s.MigrationStatus() if err != nil { @@ -615,10 +677,21 @@ func (s *Store) MigrateDown(toVersion int) error { if len(stmts) == 0 { return fmt.Errorf("migrate-down: v%d 缺少 %s 回滚", v, s.dialect.name()) } - if err := s.execInTx(stmts); err != nil { - return fmt.Errorf("migrate-down v%d (%s): %w", v, m.name, err) + tx, err := s.db.Begin() + if err != nil { + return err } - if _, err := s.db.Exec(s.q(`DELETE FROM schema_migrations WHERE version=?`), v); err != nil { + for _, q := range stmts { + if _, err := tx.Exec(s.q(q)); err != nil { + tx.Rollback() + return fmt.Errorf("migrate-down v%d (%s): %w", v, m.name, err) + } + } + if _, err := tx.Exec(s.q(`DELETE FROM schema_migrations WHERE version=?`), v); err != nil { + tx.Rollback() + return err + } + if err := tx.Commit(); err != nil { return err } } diff --git a/internal/store/migrate_atomic_test.go b/internal/store/migrate_atomic_test.go new file mode 100644 index 0000000..a89c78e --- /dev/null +++ b/internal/store/migrate_atomic_test.go @@ -0,0 +1,247 @@ +package store + +import ( + "database/sql" + "errors" + "fmt" + "os" + "path/filepath" + "strings" + "sync" + "testing" +) + +// 协作存储方案 U2（KTD8）：迁移原子性与并发正确性。 +// - DDL 与版本登记同事务：注入「DDL 已执行、版本未登记」处故障 → 整体回滚； +// - 并发迁移者竞争（SQLite _txlock=immediate / PG 咨询锁）：账本恰好一套； +// - 旧 schema（仅部分版本）打开后自动升级到最新； +// - team 档 PG 不可达：打开失败、无任何回落（R2 fail-closed 在 store 层的基线）。 + +// sqliteAt 用直接 SQL 预置一个「旧部署」库：只应用 v1 基线并登记账本。 +func sqliteAt(t *testing.T, path string) { + t.Helper() + st, err := OpenSQLite(path) + if err != nil { + t.Fatalf("seed open: %v", err) + } + if err := st.MigrateDown(1); //nolint:errcheck // 回滚到 v1 失败说明 down 链路破损，由下方状态断言兜住 + err != nil { + t.Logf("seed migrate-down: %v", err) + } + if vers, err := st.MigrationStatus(); err != nil || len(vers) == 0 { + t.Fatalf("seed status: %v %v", vers, err) + } else if vers[len(vers)-1] != 1 { + t.Fatalf("seed should stop at v1, got %v", vers) + } + st.Close() +} + +func openSQLiteAt(t *testing.T, path string) *Store { + t.Helper() + st, err := OpenSQLite(path) + if err != nil { + t.Fatal(err) + } + t.Cleanup(func() { st.Close() }) + return st +} + +// 旧 schema 升级：只有 v1 账本的库重新打开后升级到最新（双档）。 +func TestMigrateUpgradesOldSchema(t *testing.T) { + if os.Getenv("AGENT_PG_TEST_DSN") != "" { + t.Run("postgres", func(t *testing.T) { + st := openTestStore(t) + if st.Dialect() != "postgres" { + t.Skip("非 PG 档") + } + if err := st.MigrateDown(1); err != nil { + t.Fatalf("migrate-down to v1: %v", err) + } + if err := st.migrate(); err != nil { + t.Fatalf("re-migrate: %v", err) + } + assertFullLedger(t, st) + }) + } + t.Run("sqlite", func(t *testing.T) { + path := filepath.Join(t.TempDir(), "old.db") + sqliteAt(t, path) + st := openSQLiteAt(t, path) + assertFullLedger(t, st) + }) +} + +func assertFullLedger(t *testing.T, st *Store) { + t.Helper() + vers, err := st.MigrationStatus() + if err != nil { + t.Fatal(err) + } + if len(vers) != len(migrations) { + t.Fatalf("ledger %v, want %d versions", vers, len(migrations)) + } + for i, m := range migrations { + if vers[i] != m.version { + t.Fatalf("ledger[%d]=%d, want %d", i, vers[i], m.version) + } + } +} + +// DDL 与账本同事务：在「DDL 已执行、版本未登记」处注入故障 → 全部回滚， +// 不留下 v6 的表，也不留下 v6 的账本行（历史两步实现的中间态缺口）。 +func TestMigrateAtomicDDLAndLedger(t *testing.T) { + path := filepath.Join(t.TempDir(), "atomic.db") + st := openSQLiteAt(t, path) + if err := st.MigrateDown(5); err != nil { + t.Fatalf("migrate-down to v5: %v", err) + } + vers, _ := st.MigrationStatus() + if vers[len(vers)-1] != 5 { + t.Fatalf("want v5 boundary, got %v", vers) + } + boom := errors.New("crash between DDL and ledger") + st.SetMigrateFault(boom) + err := st.migrate() + if err == nil || !strings.Contains(err.Error(), "注入故障") { + t.Fatalf("want injected fault, got %v", err) + } + // 账本不含 v6；v6 的表必须不存在（同事务回滚）。 + if vers, _ = st.MigrationStatus(); vers[len(vers)-1] != 5 { + t.Fatalf("ledger advanced despite fault: %v", vers) + } + var n int + if err := st.db.QueryRow(`SELECT COUNT(*) FROM sqlite_master WHERE type='table' AND name='activity_cards'`).Scan(&n); err != nil { + t.Fatal(err) + } + if n != 0 { + t.Fatal("activity_cards exists although the version insert faulted: DDL leaked outside the transaction") + } + // 恢复后重跑：升级完成。 + if err := st.migrate(); err != nil { + t.Fatalf("retry migrate: %v", err) + } + assertFullLedger(t, st) +} + +// 并发迁移者竞争同一 SQLite 文件：_txlock=immediate + busy_timeout 串行化， +// 两边都成功，账本恰好一套（不因竞争出现重复或部分版本）。 +func TestMigrateRacingSQLite(t *testing.T) { + path := filepath.Join(t.TempDir(), "race.db") + var wg sync.WaitGroup + errs := make([]error, 4) + for i := range errs { + wg.Add(1) + go func(i int) { + defer wg.Done() + st, err := OpenSQLite(path) + if err != nil { + errs[i] = err + return + } + defer st.Close() + if vers, verr := st.MigrationStatus(); verr != nil { + errs[i] = verr + } else if len(vers) != len(migrations) { + errs[i] = fmt.Errorf("ledger=%v, want %d versions", vers, len(migrations)) + } + }(i) + } + wg.Wait() + for i, err := range errs { + if err != nil { + t.Fatalf("migrator %d: %v", i, err) + } + } + st := openSQLiteAt(t, path) + assertFullLedger(t, st) +} + +// 并发迁移者竞争同一 PG schema：会话级咨询锁串行化，账本恰好一套。 +func TestMigrateRacingPostgres(t *testing.T) { + dsn := os.Getenv("AGENT_PG_TEST_DSN") + if dsn == "" { + t.Skip("AGENT_PG_TEST_DSN 未设置：真实 PG 门禁由 CI 的必跑 job 覆盖") + } + schema := "t_race_" + strings.ToLower(newID("m")) + "_" + fmt.Sprint(os.Getpid()) + admin, err := sql.Open("pgx", dsn) + if err != nil { + t.Fatal(err) + } + if _, err := admin.Exec("CREATE SCHEMA " + schema); err != nil { + admin.Close() + t.Fatalf("create schema: %v", err) + } + t.Cleanup(func() { + admin.Exec("DROP SCHEMA IF EXISTS " + schema + " CASCADE") + admin.Close() + }) + schemaDSN := withTestSearchPath(dsn, schema) + var wg sync.WaitGroup + errs := make([]error, 3) + for i := range errs { + wg.Add(1) + go func(i int) { + defer wg.Done() + st, err := OpenPostgres(schemaDSN, filepath.Join(t.TempDir(), "artifacts")) + if err != nil { + errs[i] = err + return + } + defer st.Close() + if vers, verr := st.MigrationStatus(); verr != nil { + errs[i] = verr + } else if len(vers) != len(migrations) { + errs[i] = fmt.Errorf("ledger=%v, want %d versions", vers, len(migrations)) + } + }(i) + } + wg.Wait() + for i, err := range errs { + if err != nil { + t.Fatalf("migrator %d: %v", i, err) + } + } + // 等锁者事后看到完整账本（不因竞争出现重复/部分版本）。 + st, err := OpenPostgres(schemaDSN, filepath.Join(t.TempDir(), "artifacts")) + if err != nil { + t.Fatal(err) + } + defer st.Close() + assertFullLedger(t, st) +} + +// PG 不可达：打开失败（team 档 R2 的 store 层基线——不存在任何回落路径）。 +func TestOpenPostgresUnreachableFails(t *testing.T) { + dsn := "postgres://u:p@127.0.0.1:1/none?sslmode=disable&connect_timeout=1" + if _, err := OpenPostgres(dsn, filepath.Join(t.TempDir(), "artifacts")); err == nil { + t.Fatal("unreachable PG must fail to open") + } +} + +// 运行参数附着：URL 与键值两种 DSN 形式都正确追加 statement_timeout。 +func TestWithPGRuntimeParam(t *testing.T) { + url := withPGRuntimeParam("postgres://u:p@h:5432/db?sslmode=disable", "statement_timeout", "5000") + if want := "postgres://u:p@h:5432/db?sslmode=disable&statement_timeout=5000"; url != want { + t.Fatalf("url form: %q, want %q", url, want) + } + kv := withPGRuntimeParam("host=h port=5432 dbname=db", "statement_timeout", "5000") + if want := "host=h port=5432 dbname=db statement_timeout=5000"; kv != want { + t.Fatalf("kv form: %q, want %q", kv, want) + } + empty := withPGRuntimeParam("", "statement_timeout", "5000") + if want := "statement_timeout=5000"; empty != want { + t.Fatalf("empty dsn: %q, want %q", empty, want) + } +} + +// 连接预算有效值：零值落默认，显式值生效。 +func TestPoolOptionsEffective(t *testing.T) { + got := PoolOptions{}.effective() + if got != DefaultPGPool { + t.Fatalf("zero pool should fall back to default, got %+v", got) + } + custom := PoolOptions{MaxOpenConn: 4, MaxIdleConn: 2, ConnMaxIdleTimeSec: 60, ConnMaxLifetimeSec: 3600, QueryTimeoutSec: 5}.effective() + if custom.MaxOpenConn != 4 || custom.MaxIdleConn != 2 || custom.ConnMaxIdleTimeSec != 60 || custom.ConnMaxLifetimeSec != 3600 || custom.QueryTimeoutSec != 5 { + t.Fatalf("custom pool mangled: %+v", custom) + } +} diff --git a/internal/store/store.go b/internal/store/store.go index ae92c08..5320944 100644 --- a/internal/store/store.go +++ b/internal/store/store.go @@ -16,6 +16,8 @@ import ( "fmt" "os" "path/filepath" + "strconv" + "strings" "sync" "time" @@ -35,15 +37,46 @@ var ErrDuplicateRequest = errors.New("store: duplicate request id") // OwnerDefault 是单租户默认 owner（嵌入式档 / 未启用认证时的隐式身份）。 const OwnerDefault = "local" +// PoolOptions PG 连接预算（协作存储方案 KTD12）。零值走默认（DefaultPGPool）； +// 预算应覆盖全部 API/Worker/LISTEN/migrator 连接，100 在线用户 ≠ 100 连接。 +type PoolOptions struct { + MaxOpenConn int // 每进程最大打开连接；0=默认 16 + MaxIdleConn int // 每进程最大空闲连接；0=默认 8 + ConnMaxIdleTimeSec int // 空闲连接回收时长；0=默认 300s + ConnMaxLifetimeSec int // 连接最长生命周期；0=不限 + QueryTimeoutSec int // PG statement_timeout（秒）；0=不限 +} + +// DefaultPGPool 与历史行为一致的默认预算（16/8/300s）。 +var DefaultPGPool = PoolOptions{MaxOpenConn: 16, MaxIdleConn: 8, ConnMaxIdleTimeSec: 300} + +func (p PoolOptions) effective() PoolOptions { + out := p + if out.MaxOpenConn <= 0 { + out.MaxOpenConn = DefaultPGPool.MaxOpenConn + } + if out.MaxIdleConn <= 0 { + out.MaxIdleConn = DefaultPGPool.MaxIdleConn + } + if out.ConnMaxIdleTimeSec <= 0 { + out.ConnMaxIdleTimeSec = DefaultPGPool.ConnMaxIdleTimeSec + } + return out +} + type Store struct { db *sql.DB dialect dialect dataDir string // 工件对象目录（备份路径等诊断用） blobs blobstore.Store // 工件对象存储（本地档；多实例切共享卷/对象存储） instanceID string // 本进程标识；CreateRun 写入 runs.instance_id + pool PoolOptions // 生效的连接预算（诊断输出用） settleMu sync.Mutex settleFault error // 非空时 SettleRun 在写入前失败，用于终态提交失败测试 + + migrateMu sync.Mutex + migrateFault error // 非空时 migratePass 在「DDL 已执行、版本未登记」处失败（原子性测试） } // OpenSQLite 打开（必要时创建）嵌入式存储并执行迁移。 @@ -51,43 +84,74 @@ func OpenSQLite(path string) (*Store, error) { if err := os.MkdirAll(filepath.Dir(path), 0o755); err != nil { return nil, err } - dsn := fmt.Sprintf("file:%s?_pragma=busy_timeout(5000)&_pragma=journal_mode(WAL)&_pragma=foreign_keys(1)", path) + // _txlock=immediate：写事务在 BEGIN 即取库写锁（而非首个写语句升级）， + // 跨进程迁移/写路径依赖 busy_timeout 排队而不是中途 SQLITE_BUSY（KTD8）。 + dsn := fmt.Sprintf("file:%s?_pragma=busy_timeout(5000)&_pragma=journal_mode(WAL)&_pragma=foreign_keys(1)&_txlock=immediate", path) db, err := sql.Open("sqlite", dsn) if err != nil { return nil, err } // SQLite 单写者：串行化连接，语句级原子 + 版本化迁移事务。 db.SetMaxOpenConns(1) - return finishOpen(db, sqliteDialect{}, filepath.Join(filepath.Dir(path), "artifacts"), nil) + return finishOpen(db, sqliteDialect{}, filepath.Join(filepath.Dir(path), "artifacts"), nil, PoolOptions{}) } -// OpenPostgres 打开 PostgreSQL 档存储并执行迁移。 +// OpenPostgres 打开 PostgreSQL 档存储并执行迁移（默认连接预算）。 // DSN 例：postgres://user:pass@host:5432/agent_system?sslmode=disable // artifactsDir 为工件对象目录（多实例部署时指向共享卷；单实例默认本地）。 func OpenPostgres(dsn, artifactsDir string) (*Store, error) { - db, err := sql.Open("pgx", dsn) - if err != nil { - return nil, err - } - db.SetMaxOpenConns(16) - db.SetMaxIdleConns(8) - db.SetConnMaxIdleTime(5 * time.Minute) - return finishOpen(db, postgresDialect{}, artifactsDir, nil) + return OpenPostgresWithOptions(dsn, artifactsDir, PoolOptions{}, nil) } // OpenPostgresWithBlobs 是注入自定义对象存储的 PG 档入口（多实例共享卷/对象存储）。 func OpenPostgresWithBlobs(dsn, artifactsDir string, blobs blobstore.Store) (*Store, error) { + return OpenPostgresWithOptions(dsn, artifactsDir, PoolOptions{}, blobs) +} + +// OpenPostgresWithOptions 按 KTD12 连接预算打开 PG 档；QueryTimeoutSec>0 时 +// 以 statement_timeout 运行参数附着到 DSN（对 URL 与键值两种形式均生效）。 +func OpenPostgresWithOptions(dsn, artifactsDir string, pool PoolOptions, blobs blobstore.Store) (*Store, error) { + pool = pool.effective() + if pool.QueryTimeoutSec > 0 { + dsn = withPGRuntimeParam(dsn, "statement_timeout", strconv.Itoa(pool.QueryTimeoutSec*1000)) + } db, err := sql.Open("pgx", dsn) if err != nil { return nil, err } - db.SetMaxOpenConns(16) - db.SetMaxIdleConns(8) - db.SetConnMaxIdleTime(5 * time.Minute) - return finishOpen(db, postgresDialect{}, artifactsDir, blobs) + db.SetMaxOpenConns(pool.MaxOpenConn) + db.SetMaxIdleConns(pool.MaxIdleConn) + db.SetConnMaxIdleTime(time.Duration(pool.ConnMaxIdleTimeSec) * time.Second) + if pool.ConnMaxLifetimeSec > 0 { + db.SetConnMaxLifetime(time.Duration(pool.ConnMaxLifetimeSec) * time.Second) + } + return finishOpen(db, postgresDialect{}, artifactsDir, blobs, pool) } -func finishOpen(db *sql.DB, d dialect, artifactsDir string, blobs blobstore.Store) (*Store, error) { +// withPGRuntimeParam 向 DSN 追加运行时参数（libpq 语义：URL 查询串或键值对）。 +// 不改写既有参数；值中不含分隔符时安全（statement_timeout/search_path 均满足）。 +func withPGRuntimeParam(dsn, key, value string) string { + return AppendPGRuntimeParam(dsn, key, value) +} + +// AppendPGRuntimeParam 导出形态：测试与其他包（如 runmgr 的 schema 隔离） +// 共用同一 DSN 拼接规则，避免各自盲拼 "&key=value" 造出非法库名。 +func AppendPGRuntimeParam(dsn, key, value string) string { + kv := key + "=" + value + if !strings.Contains(dsn, "://") { + if strings.TrimSpace(dsn) == "" { + return kv + } + return dsn + " " + kv + } + sep := "?" + if strings.Contains(dsn, "?") { + sep = "&" + } + return dsn + sep + kv +} + +func finishOpen(db *sql.DB, d dialect, artifactsDir string, blobs blobstore.Store, pool PoolOptions) (*Store, error) { if blobs == nil { var err error blobs, err = blobstore.NewLocal(artifactsDir) @@ -96,7 +160,7 @@ func finishOpen(db *sql.DB, d dialect, artifactsDir string, blobs blobstore.Stor return nil, err } } - s := &Store{db: db, dialect: d, dataDir: artifactsDir, blobs: blobs} + s := &Store{db: db, dialect: d, dataDir: artifactsDir, blobs: blobs, pool: pool} if err := s.migrate(); err != nil { db.Close() return nil, err @@ -121,9 +185,28 @@ func (s *Store) SetSettleFault(err error) { s.settleMu.Unlock() } +// SetMigrateFault 让下一次 migratePass 在首个待应用迁移的「DDL 已执行、版本 +// 未登记」位置失败一次（nil 恢复）。测试专用：验证 DDL 与账本同事务回滚。 +func (s *Store) SetMigrateFault(err error) { + s.migrateMu.Lock() + s.migrateFault = err + s.migrateMu.Unlock() +} + +func (s *Store) takeMigrateFault() error { + s.migrateMu.Lock() + defer s.migrateMu.Unlock() + err := s.migrateFault + s.migrateFault = nil + return err +} + // Dialect 返回当前存储档名（sqlite | postgres；诊断/健康检查用）。 func (s *Store) Dialect() string { return s.dialect.name() } +// Pool 返回生效的连接预算（启动诊断；脱敏，不含 DSN）。 +func (s *Store) Pool() PoolOptions { return s.pool } + // DataDir 返回工件对象目录。 func (s *Store) DataDir() string { return s.dataDir } diff --git a/internal/store/store_test.go b/internal/store/store_test.go index ef307e8..d5d2e32 100644 --- a/internal/store/store_test.go +++ b/internal/store/store_test.go @@ -31,6 +31,12 @@ func openTestStore(t *testing.T) *Store { if err != nil { t.Fatal(err) } + // pg_trgm 预装到 public（幂等）：v4 迁移的 CREATE EXTENSION IF NOT EXISTS + // 会装进当时 search_path 的首个 schema（= 用例 schema），随 DROP SCHEMA + // CASCADE 被连带删除——并行用例间形成安装/删除竞态，gin_trgm_ops 随机 + // 缺失。预装到 public 后 v4 的 CREATE EXTENSION 恒为 no-op，操作符经 + // search_path 的 public 段稳定可见（与 svc.sh/CI 预装约定一致）。 + ensurePGTrgm(t, admin) if _, err := admin.Exec("CREATE SCHEMA " + schema); err != nil { admin.Close() t.Fatalf("create schema: %v", err) @@ -47,6 +53,19 @@ func openTestStore(t *testing.T) *Store { return st } +// ensurePGTrgm 在测试库 public 幂等安装 pg_trgm；并发首装的目录唯一冲突 +// 重试一次后放弃（下一个用例会再命中幂等分支）。 +func ensurePGTrgm(t *testing.T, admin *sql.DB) { + t.Helper() + for attempt := 0; attempt < 2; attempt++ { + if _, err := admin.Exec(`CREATE EXTENSION IF NOT EXISTS pg_trgm`); err == nil { + return + } else if attempt == 1 { + t.Fatalf("preinstall pg_trgm: %v", err) + } + } +} + // withTestSearchPath 把用例 schema 放到 search_path 最前，并保留 public， // 以便看到已安装在 public 的 pg_trgm。svc.sh 给出的是无查询串的 URL； // 键值对 DSN（不含 ://）用空格追加。