weichen
5d5cef9fcd
ci / vet-and-test (pull_request) Successful in 56s
Why: plan U6 vector slice (KTD11 correctness step) - same-dimension vectors from different models must never mix, and PG tiers should run filtered exact top-k in the database instead of scanning every row in-process. Evidence: full suite green on SQLite and real PostgreSQL 18.6 with pgvector (35 packages each tier) - pgvector_test covers model-identity and dimension isolation, update/delete atomicity (BLOB and vec_pg swap together), shared top-5 with score monotonicity, Recall@10 = exact-scan parity, deny-all predicate leakage; storage-migrate tests green after excluding the derived vec_pg column. Impact: v25 migration adds model/content_hash columns and (PG, best-effort DO block) the pgvector extension plus vec_pg column - absence degrades to in-process scan; UpsertKnowledgeVectorExt dual-writes BLOB (cross-tier authority) and vec_pg; SearchKnowledgeVectorsModel/SharedModel filter by model identity and use SQL exact top-k (ORDER BY vec_pg <=> q) with automatic degradation; embedders expose Model() and the rag pipeline threads it through indexing and both retrieve paths; storage-migrate excludes vec_pg as a derived column. Refs: proposals/2026-10-11-collaboration-storage-and-distribution.md U6 slice 3, KTD11; R9.
262 lines
9.9 KiB
Go
262 lines
9.9 KiB
Go
// Package store 是持久化层唯一入口（存储档接口，设计 4.1/D5/D16）。
|
||
// 两档实现共用同一方法面：
|
||
// - 嵌入式档 SQLite（默认：单文件 + 本地对象目录，零外部服务）；
|
||
// - PostgreSQL 档（多用户/多实例；DSN 配置，pgx 纯 Go 驱动，静态构建兼容）。
|
||
//
|
||
// 方言差异（占位符、UPSERT、全文检索、备份、咨询锁）收敛在 dialect 与方言文件；
|
||
// 业务层只依赖本包公开方法。schema 演进走版本化迁移（migrate.go，双方言、可回滚）。
|
||
// 租户硬边界：租户域查询一律带 owner 过滤，跨租户访问返回 ErrNotFound。
|
||
package store
|
||
|
||
import (
|
||
"context"
|
||
"database/sql"
|
||
"encoding/json"
|
||
"errors"
|
||
"fmt"
|
||
"os"
|
||
"path/filepath"
|
||
"strconv"
|
||
"strings"
|
||
"sync"
|
||
"time"
|
||
|
||
"agent-system-backend/internal/blobstore"
|
||
|
||
_ "github.com/jackc/pgx/v5/stdlib" // PG 档驱动（注册名 "pgx"）
|
||
_ "modernc.org/sqlite" // 嵌入式档驱动（注册名 "sqlite"，CGO-free）
|
||
)
|
||
|
||
var ErrNotFound = errors.New("store: not found")
|
||
var ErrConflict = errors.New("store: conflict")
|
||
|
||
// ErrDuplicateRequest 表示同一会话的 request_id 已经落库。
|
||
// 调用方应返回已有消息，不要映射成 409：客户端把 409 当成会话忙并重新入队。
|
||
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 已执行、版本未登记」处失败（原子性测试）
|
||
|
||
vecPGOnce sync.Once // pgvector 列探测（U6-3：一次性缓存）
|
||
vecPG bool
|
||
}
|
||
|
||
// OpenSQLite 打开（必要时创建）嵌入式存储并执行迁移。
|
||
func OpenSQLite(path string) (*Store, error) {
|
||
if err := os.MkdirAll(filepath.Dir(path), 0o755); err != nil {
|
||
return nil, err
|
||
}
|
||
// _txlock=immediate：写事务在 BEGIN 即取库写锁（而非首个写语句升级），
|
||
// 跨进程迁移/写路径依赖 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)
|
||
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, PoolOptions{})
|
||
}
|
||
|
||
// OpenPostgres 打开 PostgreSQL 档存储并执行迁移（默认连接预算）。
|
||
// DSN 例：postgres://user:pass@host:5432/agent_system?sslmode=disable
|
||
// artifactsDir 为工件对象目录（多实例部署时指向共享卷；单实例默认本地）。
|
||
func OpenPostgres(dsn, artifactsDir string) (*Store, error) {
|
||
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(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)
|
||
}
|
||
|
||
// 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)
|
||
if err != nil {
|
||
db.Close()
|
||
return nil, err
|
||
}
|
||
}
|
||
s := &Store{db: db, dialect: d, dataDir: artifactsDir, blobs: blobs, pool: pool}
|
||
if err := s.migrate(); err != nil {
|
||
db.Close()
|
||
return nil, err
|
||
}
|
||
if err := os.MkdirAll(artifactsDir, 0o755); err != nil {
|
||
db.Close()
|
||
return nil, err
|
||
}
|
||
return s, nil
|
||
}
|
||
|
||
func (s *Store) Close() error { return s.db.Close() }
|
||
|
||
// SetInstanceID 记录本进程标识。之后 CreateRun 把它写入 runs.instance_id。
|
||
// 只在开始接请求前调用一次。
|
||
func (s *Store) SetInstanceID(id string) { s.instanceID = id }
|
||
|
||
// SetSettleFault 让下一次 SettleRun 在写入前失败。nil 表示恢复正常。
|
||
func (s *Store) SetSettleFault(err error) {
|
||
s.settleMu.Lock()
|
||
s.settleFault = err
|
||
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 }
|
||
|
||
// DB 暴露底层连接（运维工具专用：storage-migrate 的控制表与
|
||
// information_schema 查询）。业务代码一律走本包公开方法。
|
||
func (s *Store) DB() *sql.DB { return s.db }
|
||
|
||
// DataDir 返回工件对象目录。
|
||
func (s *Store) DataDir() string { return s.dataDir }
|
||
|
||
// BlobStore 返回工件对象存储（诊断/运维）。
|
||
func (s *Store) BlobStore() blobstore.Store { return s.blobs }
|
||
|
||
// NotifyPG 发送 PostgreSQL NOTIFY（PG 档事件扇出用；SQLite 档返回未执行）。
|
||
func (s *Store) NotifyPG(ctx context.Context, channel, payload string) (bool, error) {
|
||
if s.dialect.name() != "postgres" {
|
||
return false, nil
|
||
}
|
||
_, err := s.db.ExecContext(ctx, `SELECT pg_notify($1, $2)`, channel, payload)
|
||
return true, err
|
||
}
|
||
|
||
// q 按方言重绑占位符（查询统一用 ? 书写，PG 执行前转 $n）。
|
||
// 本包 SQL 不含字符串字面量 "?"，重绑安全。
|
||
func (s *Store) q(query string) string { return s.dialect.rebind(query) }
|
||
|
||
// Backup 生成一致性备份文件。SQLite 档用 VACUUM INTO（在线安全，含 WAL 内容）；
|
||
// PG 档备份走 pg_dump / WAL 归档（运维路径，见 usage 指南），此处返回指引错误。
|
||
func (s *Store) Backup(destPath string) error {
|
||
if s.dialect.name() != "sqlite" {
|
||
return errors.New("store: PG 档备份走 pg_dump/WAL 归档（见 docs/guides/usage.md 存储档一节）")
|
||
}
|
||
if _, err := s.db.Exec(`VACUUM INTO ?`, destPath); err != nil {
|
||
return fmt.Errorf("store: backup: %w", err)
|
||
}
|
||
return nil
|
||
}
|
||
|
||
func nowISO() string { return time.Now().UTC().Format(time.RFC3339Nano) }
|
||
|
||
func marshalJSON(v any) string {
|
||
b, _ := json.Marshal(v)
|
||
return string(b)
|
||
}
|
||
|
||
func unmarshalJSON[T any](raw string) (T, error) {
|
||
var out T
|
||
err := json.Unmarshal([]byte(raw), &out)
|
||
return out, err
|
||
}
|