fix: KB 级联删「事务化」—— 失败不再留不可删孤儿(T4.F)

原为 best-effort:三库删失败只 log、MinIO 删错误全吞、PG 照删 → 删一半失败即在
向量/全文/图谱/MinIO 留下「PG 无记录、连 file_id 都查不到」的不可删孤儿。

改为「类事务」(跨库 2PC 不可行,退而求其次:不留不可恢复孤儿 + 失败可见可重试):
- milvus.deleteByFile / bleve.deleteDoc / blob.Delete 改返回 error(原 void 吞错)
- rag.DeleteDoc 三库全试一遍(最大化清理)+ 聚合错误(原只回 Neo4j 的错);三库删幂等
- gateway KbDeleteDoc 失败闭合:先删依赖存储(三库→MinIO)、PG 最后删;
  任一存储删失败 → 不删 PG、返 502「未删除请重试」(保留 file_id 供幂等重试)
- 语义翻转:从「总能从列表删掉但留孤儿」→「有孤儿风险就不删、报错可重试」

live:杀 mcp-go→删→502+文档保留;mcp-go 活→删→200+清空(清场僵尸进程后验证)

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
This commit is contained in:
Blizzard
2026-07-06 16:32:25 +08:00
parent 55d50417a9
commit 53f7e172c3
6 changed files with 44 additions and 22 deletions
+4 -1
View File
@@ -193,7 +193,10 @@ RBAC 未做,暂以单管理员账号代理;概览口径必须是**系统级*
### [ ] T4.F 健壮性 / 安全 / 性能收口(多为 S,可穿插着做) ### [ ] T4.F 健壮性 / 安全 / 性能收口(多为 S,可穿插着做)
- [ ] 关键 DB 写失败上浮 5xx(现 best-effort 返 200,前端无感,task_handler.go:69 等)| M - [ ] 关键 DB 写失败上浮 5xx(现 best-effort 返 200,前端无感,task_handler.go:69 等)| M
- [ ] KB 级联删事务化(kb.go:172 三库删一半失败仍删 PG → 不一致)| M - [x] KB 级联删事务化」✅(2026-07-06)—— 类事务:先删依赖存储(三库/MinIO)、PG 记录**最后删**
任一存储删失败→不删 PG、报错可重试(三库删幂等),杜绝「PG 删了却在三库/MinIO 留不可删孤儿」。
milvus/bleve.deleteDoc + blob.Delete 改返 errorrag.DeleteDoc 三库全试+聚合错误(原只回 Neo4j 的错)。
**live:杀 mcp-go→删→502「未删除请重试」+文档保留;mcp-go 活→删→200+清空**(清场僵尸进程后验证)。
- [ ] 预算硬顶兜底(budget.go ≤0 即无限,单任务可无限烧)| M - [ ] 预算硬顶兜底(budget.go ≤0 即无限,单任务可无限烧)| M
- [ ] 审批 checkpoint 落盘失败重试(compose_compiler.go:323 现只 log → 任务永卡 waiting| M - [ ] 审批 checkpoint 落盘失败重试(compose_compiler.go:323 现只 log → 任务永卡 waiting| M
- [~] 安全收口(部分): - [~] 安全收口(部分):
+3 -3
View File
@@ -69,7 +69,7 @@ func (s *Store) Get(ctx context.Context, key string) (string, error) {
return string(b), nil return string(b), nil
} }
// Delete 删除对象best-effort)。 // Delete 删除对象;返回错误供级联删按需失败闭合(其它 best-effort 调用点可忽略返回值)。
func (s *Store) Delete(ctx context.Context, key string) { func (s *Store) Delete(ctx context.Context, key string) error {
_ = s.cli.RemoveObject(ctx, s.bucket, key, minio.RemoveObjectOptions{}) return s.cli.RemoveObject(ctx, s.bucket, key, minio.RemoveObjectOptions{})
} }
+14 -6
View File
@@ -168,22 +168,30 @@ func (h *Handler) KbDeleteDoc(c *gin.Context) {
return return
} }
scoped := d.Owner + "/" + d.KB scoped := d.Owner + "/" + d.KB
// ① 三库按 file_id 级联删(失败不中断,避免半残;记录降级)。 // 「类事务」删除:先删依赖存储(三库/MinIO),PG 记录**最后删**。任一存储删失败 → 不删 PG、
// 保留记录报错让用户重试(三库/MinIO 删均幂等);杜绝「PG 删了却在三库/MinIO 留下再也删不掉的孤儿」。
// ① 三库按 file_id 级联删。失败 → 保留 PG 可重试。
if res, e := h.bus.CallTool(c.Request.Context(), contract.ToolSubjectGo("kb_delete"), if res, e := h.bus.CallTool(c.Request.Context(), contract.ToolSubjectGo("kb_delete"),
&contract.ToolCall{Tool: "kb_delete", Args: map[string]any{"kb": scoped, "file_id": d.ID}}); e != nil || res == nil || !res.OK { &contract.ToolCall{Tool: "kb_delete", Args: map[string]any{"kb": scoped, "file_id": d.ID}}); e != nil || res == nil || !res.OK {
msg := "kb_delete 无响应" msg := "kb_delete 无响应"
if e != nil { if e != nil {
msg = e.Error() msg = e.Error()
} else if res != nil { } else if res != nil && res.Error != "" {
msg = res.Error msg = res.Error
} }
log.Printf("[gateway] 三库删除降级 id=%s: %s", d.ID, msg) log.Printf("[gateway] 三库删除失败,保留记录待重试 id=%s: %s", d.ID, msg)
c.JSON(http.StatusBadGateway, gin.H{"error": "存储删除失败,未删除,请稍后重试:" + msg})
return
} }
// ② MinIO 原文。 // ② MinIO 原文。失败 → 同样保留 PG 可重试(三库已幂等清理,重试补删 MinIO+PG)。
if d.ObjectKey != "" && h.blob.Ready() { if d.ObjectKey != "" && h.blob.Ready() {
h.blob.Delete(c.Request.Context(), d.ObjectKey) if e := h.blob.Delete(c.Request.Context(), d.ObjectKey); e != nil {
log.Printf("[gateway] MinIO 删除失败,保留记录待重试 id=%s: %v", d.ID, e)
c.JSON(http.StatusBadGateway, gin.H{"error": "原文删除失败,未删除,请稍后重试"})
return
}
} }
// ③ PG 元数据 + 双链。 // ③ PG 元数据 + 双链(最后删;到此依赖存储已清,PG 删成即整体一致)
if e := h.db.DeleteDocByID(c.Request.Context(), owner, d.KB, d.ID); e != nil { if e := h.db.DeleteDocByID(c.Request.Context(), owner, d.KB, d.ID); e != nil {
c.JSON(http.StatusBadGateway, gin.H{"error": e.Error()}) c.JSON(http.StatusBadGateway, gin.H{"error": e.Error()})
return return
+4 -4
View File
@@ -101,9 +101,9 @@ func (b *bleveStore) index(kb, doc string, texts []string) error {
} }
// deleteDoc 删除某 (kb, doc) 的全部全文块(笔记重入库前清旧块)。 // deleteDoc 删除某 (kb, doc) 的全部全文块(笔记重入库前清旧块)。
func (b *bleveStore) deleteDoc(kb, doc string) { func (b *bleveStore) deleteDoc(kb, doc string) error {
if !b.ready() || doc == "" { if !b.ready() || doc == "" {
return return nil
} }
kq := bleve.NewTermQuery(kb) kq := bleve.NewTermQuery(kb)
kq.SetField("kb") kq.SetField("kb")
@@ -113,13 +113,13 @@ func (b *bleveStore) deleteDoc(kb, doc string) {
req.Size = 1000 req.Size = 1000
res, err := b.idx.Search(req) res, err := b.idx.Search(req)
if err != nil { if err != nil {
return return err
} }
batch := b.idx.NewBatch() batch := b.idx.NewBatch()
for _, h := range res.Hits { for _, h := range res.Hits {
batch.Delete(h.ID) batch.Delete(h.ID)
} }
_ = b.idx.Batch(batch) return b.idx.Batch(batch)
} }
// search 全文检索(可按 kb 过滤),返回 BM25 排序的命中。 // search 全文检索(可按 kb 过滤),返回 BM25 排序的命中。
+5 -3
View File
@@ -116,14 +116,16 @@ func (m *milvusStore) deleteDoc(ctx context.Context, kb, doc string, dim int) {
} }
// deleteByFile 按 file_id 删某文档的全部向量块(级联删用,无需 dim)。集合不存在/未加载时静默忽略。 // deleteByFile 按 file_id 删某文档的全部向量块(级联删用,无需 dim)。集合不存在/未加载时静默忽略。
func (m *milvusStore) deleteByFile(ctx context.Context, kb, fileID string) { func (m *milvusStore) deleteByFile(ctx context.Context, kb, fileID string) error {
if fileID == "" { if fileID == "" {
return return nil
} }
expr := fmt.Sprintf("kb == %q && doc == %q", kb, fileID) expr := fmt.Sprintf("kb == %q && doc == %q", kb, fileID)
if err := m.cli.Delete(ctx, collection, "", expr); err != nil { if err := m.cli.Delete(ctx, collection, "", expr); err != nil {
log.Printf("[rag] 按 file_id 删除向量失败(忽略): %v", err) log.Printf("[rag] 按 file_id 删除向量失败: %v", err)
return err
} }
return nil
} }
// insert 写入若干 (kb, doc, text, vector)。 // insert 写入若干 (kb, doc, text, vector)。
+14 -5
View File
@@ -5,6 +5,7 @@ package rag
import ( import (
"context" "context"
"errors" "errors"
"fmt"
"log" "log"
"os" "os"
"strconv" "strconv"
@@ -253,16 +254,24 @@ func (e *Engine) Search(ctx context.Context, kb, query string, topK int) ([]Hit,
} }
// DeleteDoc 按 file_id 级联删某文档在三库的痕迹:Milvus 向量块 + Bleve 全文块 + Neo4j 关系。 // DeleteDoc 按 file_id 级联删某文档在三库的痕迹:Milvus 向量块 + Bleve 全文块 + Neo4j 关系。
// Milvus/Bleve 尽力删(void),Neo4j 返回错误供上层感知;任一失败不阻断其余。 // DeleteDoc 按 file_id 级联删三库痕迹。三库**全试一遍**(最大化清理,不因一处失败漏删另两处),
// 聚合各库错误上返。三库删均幂等(删不存在=no-op),故上层可安全重试。
func (e *Engine) DeleteDoc(ctx context.Context, kb, fileID string) error { func (e *Engine) DeleteDoc(ctx context.Context, kb, fileID string) error {
if fileID == "" { if fileID == "" {
return errors.New("file_id 必填") return errors.New("file_id 必填")
} }
e.mv.deleteByFile(ctx, kb, fileID) var errs []error
e.bleve.deleteDoc(kb, fileID) if err := e.mv.deleteByFile(ctx, kb, fileID); err != nil {
errs = append(errs, fmt.Errorf("向量: %w", err))
}
if err := e.bleve.deleteDoc(kb, fileID); err != nil {
errs = append(errs, fmt.Errorf("全文: %w", err))
}
if err := e.graph.deleteByFile(ctx, kb, fileID); err != nil { if err := e.graph.deleteByFile(ctx, kb, fileID); err != nil {
log.Printf("[rag] 图谱按 file_id 删除失败: %v", err) errs = append(errs, fmt.Errorf("图谱: %w", err))
return err }
if len(errs) > 0 {
return fmt.Errorf("级联删部分失败(幂等,可重试): %w", errors.Join(errs...))
} }
log.Printf("[rag] 已删除文档痕迹 kb=%s file_id=%s(向量/全文/图谱)", kb, fileID) log.Printf("[rag] 已删除文档痕迹 kb=%s file_id=%s(向量/全文/图谱)", kb, fileID)
return nil return nil