package main import ( "crypto/sha256" "encoding/binary" "encoding/hex" "encoding/json" "log" "os" "path/filepath" "sync" "time" "git.ycl.pub/yl32940/TsDB2/DB" ) // ===== 审计结论缓存库 ===== // // 为什么要落盘 // ------------ // 全库审计在千万行级别要十几秒 CPU、数 GB 内存 (见 indexcheck.go 里那几张指纹集), // 而结论只在库内容真的变了以后才需要重算。把结论存下来, 下次打开同一个库直接取, // 既省掉一遍全库扫描, 也不用再把中间过程堆在内存里。 // // 存在哪 // ------ // 被分析库旁边的独立库文件 ".analyzer", 用**引擎自己的 ORM** 组织: // audit_cache 每个指纹一行 —— 摘要 (判决 / 页账 / 每表每索引统计 / 说明) 的 JSON // audit_issue 索引内容问题明细, 一行一条 → 可按页取, 不必整份读进内存 // audit_unacc 孤儿页明细, 一行一条 // 明细拆表就是为了分页: 库里出了问题可能有成千上万条, 一次性塞给前端既慢又占内存。 // // 什么时候算"同一个库" // -------------------- // 指纹 = sha256(文件大小 + 头部页 4096 字节)。头部里有 ModifyTime / TxnID / FreeStart / // NextFreePage / TablePage 等字段, 引擎每写一次都会动其中之一, 所以指纹一致基本等价于 // "文件没被改过"; 不一致就老老实实重算 —— 宁可多算一次, 也不能拿旧结论糊弄人。 // // 容错 // ---- // 缓存是纯粹的加速层。任何一步出错 (文件被占 / 引擎报错 / 结构不兼容) 都只记日志并 // 退回"跑一遍全量审计", 绝不影响结论正确性。 // auditCacheRow 一次审计的摘要。Payload 是 AuditResult 的 JSON (去掉了明细列表, // 明细在下面两张表里)。 type auditCacheRow struct { DB.BaseModel FP string `vrtx:"maxlen:80,index"` // 库指纹 Updated int64 // 写入时间 (Unix 秒) Payload []byte // 动态列: 存 JSON 本体 } // auditIssueRow 索引内容问题明细, 供分页查看。 type auditIssueRow struct { DB.BaseModel FP string `vrtx:"maxlen:80,index"` Table string `vrtx:"maxlen:64"` Index string `vrtx:"maxlen:64"` IType string `vrtx:"maxlen:24"` Kind string `vrtx:"maxlen:32"` Page uint32 At int Detail string `vrtx:"maxlen:256"` } // auditUnaccRow 孤儿页明细 (既不属于任何结构、也不在空闲链上的页)。 type auditUnaccRow struct { DB.BaseModel FP string `vrtx:"maxlen:80,index"` Page uint32 } var ( storeMu sync.Mutex // 引擎非线程安全 —— 缓存库的所有访问都串行化 cacheFP string // 当前库指纹 (由 openH 设置) ) // cachePath 缓存库路径。可用 TSDB_ANALYZER_CACHE 指定到别处 (比如非持久盘)。 func cachePath(dbPath string) string { if v := os.Getenv("TSDB_ANALYZER_CACHE"); v != "" { return v } return dbPath + ".analyzer" } // setCacheFP 记录当前库指纹 (供分页查询定位缓存行)。 func setCacheFP(fp string) { cacheMu.Lock() cacheFP = fp cacheMu.Unlock() } // fingerprint 当前库的指纹。必须在头部页已加载 (rHdr) 之后调用。 func fingerprint() string { if dbF == "" || len(hdr) == 0 { return "" } h := sha256.New() var sz [8]byte if fi, err := os.Stat(dbF); err == nil { binary.LittleEndian.PutUint64(sz[:], uint64(fi.Size())) } h.Write(sz[:]) h.Write(hdr[:]) return hex.EncodeToString(h.Sum(nil)[:20]) } // payloadOf 复制一份可落盘的审计结果: 明细列表与空闲页列表都摘出去。 // 注意 Free / Reach 是**指针**, 必须先复制再改, 不能就地清空 —— 那份结果可能还在用。 func payloadOf(a *AuditResult) *AuditResult { b := *a b.IdxIssues = nil if b.Free != nil { f := *b.Free f.Pages = nil // 空闲页由 /api/free-pages 现走链取, 不在这份里重复存 b.Free = &f } if b.Reach != nil { r := *b.Reach r.UnaccountedIDs = nil b.Reach = &r } return &b } // cacheOpen 打开缓存库并关掉自动提交 (由我们显式 Commit)。 func cacheOpen() (*DB.DB, error) { db, err := DB.Open(cachePath(dbF)) if err != nil { return nil, err } db.SetAutoCommitDelay(0) return db, nil } func removeCache() { p := cachePath(dbF) _ = os.Remove(p) _ = os.Remove(p + ".wal") } // storeSave 把一份**完整**的审计结论落盘。 // 整库重写: 缓存只服务于一个库路径, 全量重建比"按指纹删旧行"简单得多, 也不会留下 // 历史指纹的垃圾行。 func storeSave(fp string, a *AuditResult) { if fp == "" || a == nil || a.IdxPending || a.Reach == nil { return } payload, err := json.Marshal(payloadOf(a)) if err != nil { log.Printf("[cache] 序列化失败: %v", err) return } if err := doSave(fp, payload, a); err != nil { // 结构不兼容 / 文件损坏: 删掉重建一次, 还不行就放弃缓存。 removeCache() if err2 := doSave(fp, payload, a); err2 != nil { log.Printf("[cache] 写入失败, 本次不缓存: %v", err2) } } } func doSave(fp string, payload []byte, a *AuditResult) error { storeMu.Lock() defer storeMu.Unlock() removeCache() // 全量重建 db, err := cacheOpen() if err != nil { return err } defer db.Close() meta, err := DB.UseTable[auditCacheRow](db, "audit_cache") if err != nil { return err } iss, err := DB.UseTable[auditIssueRow](db, "audit_issue") if err != nil { return err } una, err := DB.UseTable[auditUnaccRow](db, "audit_unacc") if err != nil { return err } if err := meta.Insert(&auditCacheRow{FP: fp, Updated: time.Now().Unix(), Payload: payload}); err != nil { return err } for i := range a.IdxIssues { is := &a.IdxIssues[i] if err := iss.Insert(&auditIssueRow{ FP: fp, Table: is.Table, Index: is.Index, IType: is.IType, Kind: is.Kind, Page: is.Page, At: is.At, Detail: clip(is.Detail, 240), }); err != nil { return err } } for _, p := range a.Reach.UnaccountedIDs { if err := una.Insert(&auditUnaccRow{FP: fp, Page: p}); err != nil { return err } } return db.Commit() } // storeLoad 尝试取回同一指纹的审计结论。命中返回 (结果, true)。 // gen 为调用时刻的切库代数 —— 期间用户换库就别往回装了。 func storeLoad(fp string, gen uint64) (*AuditResult, bool) { if fp == "" { return nil, false } a, err := doLoad(fp) if err != nil || a == nil { if err != nil { log.Printf("[cache] 读取失败, 转为全量分析: %v", err) } return nil, false } a.IdxPending = false _ = gen return a, true } func doLoad(fp string) (*AuditResult, error) { storeMu.Lock() defer storeMu.Unlock() if _, err := os.Stat(cachePath(dbF)); err != nil { return nil, nil // 没有缓存文件 = 未命中, 不是错误 } db, err := cacheOpen() if err != nil { return nil, err } defer db.Close() meta, err := DB.UseTable[auditCacheRow](db, "audit_cache") if err != nil { return nil, err } rows, err := meta.Where("FP", "=", fp).Find() if err != nil { return nil, err } if len(rows) == 0 { return nil, nil } a := &AuditResult{} if err := json.Unmarshal(rows[0].Payload, a); err != nil { return nil, err } if a.Reach == nil { a.Reach = &ReachResult{ByRole: map[string]int{}, UnaccountedByType: map[string]int{}} } if a.Free == nil { a.Free = &FreeChainInfo{} } if iss, err := DB.UseTable[auditIssueRow](db, "audit_issue"); err == nil { if rs, err := iss.Where("FP", "=", fp).Find(); err == nil { for _, r := range rs { a.IdxIssues = append(a.IdxIssues, IndexIssue{ Table: r.Table, Index: r.Index, IType: r.IType, Kind: r.Kind, Page: r.Page, At: r.At, Detail: r.Detail, }) } } } if una, err := DB.UseTable[auditUnaccRow](db, "audit_unacc"); err == nil { if rs, err := una.Where("FP", "=", fp).Find(); err == nil { for _, r := range rs { a.Reach.UnaccountedIDs = append(a.Reach.UnaccountedIDs, r.Page) } } } return a, nil } // ===== 明细分页查询 (从缓存库里按页取, 不整份读进内存) ===== // cachePageIssues 分页取索引问题明细。第二个返回值是总数。 func cachePageIssues(offset, limit int) ([]IndexIssue, int, error) { storeMu.Lock() defer storeMu.Unlock() cacheMu.Lock() fp := cacheFP cacheMu.Unlock() if fp == "" { return nil, 0, os.ErrNotExist } db, err := cacheOpen() if err != nil { return nil, 0, err } defer db.Close() t, err := DB.UseTable[auditIssueRow](db, "audit_issue") if err != nil { return nil, 0, err } total, err := t.Where("FP", "=", fp).Count() if err != nil { return nil, 0, err } rows, err := t.Where("FP", "=", fp).OrderBy("_id", false).Offset(offset).Limit(limit).Find() if err != nil { return nil, 0, err } out := make([]IndexIssue, 0, len(rows)) for _, r := range rows { out = append(out, IndexIssue{ Table: r.Table, Index: r.Index, IType: r.IType, Kind: r.Kind, Page: r.Page, At: r.At, Detail: r.Detail, }) } return out, int(total), nil } // cachePageUnacc 分页取孤儿页明细。第二个返回值是总数。 func cachePageUnacc(offset, limit int) ([]uint32, int, error) { storeMu.Lock() defer storeMu.Unlock() cacheMu.Lock() fp := cacheFP cacheMu.Unlock() if fp == "" { return nil, 0, os.ErrNotExist } db, err := cacheOpen() if err != nil { return nil, 0, err } defer db.Close() t, err := DB.UseTable[auditUnaccRow](db, "audit_unacc") if err != nil { return nil, 0, err } total, err := t.Where("FP", "=", fp).Count() if err != nil { return nil, 0, err } rows, err := t.Where("FP", "=", fp).OrderBy("_id", false).Offset(offset).Limit(limit).Find() if err != nil { return nil, 0, err } out := make([]uint32, 0, len(rows)) for _, r := range rows { out = append(out, r.Page) } return out, int(total), nil } // clip 把字符串截到 n 字节以内 (按 rune 边界, 不切出半个汉字)。 func clip(s string, n int) string { if len(s) <= n { return s } rs := []rune(s) for len(string(rs)) > n { rs = rs[:len(rs)-1] } return string(rs) } // cacheFileInfo 缓存库的位置与大小, 仅供日志/调试。 func cacheFileInfo() (string, int64) { p := cachePath(dbF) if dbF == "" { return "", 0 } if fi, err := os.Stat(p); err == nil { return filepath.Clean(p), fi.Size() } return filepath.Clean(p), 0 }