From 53f7e172c3dae8cd6b850b1db85a14e3f88540ed Mon Sep 17 00:00:00 2001 From: Blizzard Date: Mon, 6 Jul 2026 16:32:25 +0800 Subject: [PATCH] =?UTF-8?q?fix:=20KB=20=E7=BA=A7=E8=81=94=E5=88=A0?= =?UTF-8?q?=E3=80=8C=E4=BA=8B=E5=8A=A1=E5=8C=96=E3=80=8D=E2=80=94=E2=80=94?= =?UTF-8?q?=20=E5=A4=B1=E8=B4=A5=E4=B8=8D=E5=86=8D=E7=95=99=E4=B8=8D?= =?UTF-8?q?=E5=8F=AF=E5=88=A0=E5=AD=A4=E5=84=BF=EF=BC=88T4.F=EF=BC=89?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 原为 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 --- DEPTH_ROADMAP.md | 5 ++++- sundynix-gateway/internal/blob/minio.go | 6 +++--- sundynix-gateway/internal/handler/kb.go | 20 ++++++++++++++------ sundynix-mcp-go/internal/rag/bleve.go | 8 ++++---- sundynix-mcp-go/internal/rag/milvus.go | 8 +++++--- sundynix-mcp-go/internal/rag/rag.go | 19 ++++++++++++++----- 6 files changed, 44 insertions(+), 22 deletions(-) diff --git a/DEPTH_ROADMAP.md b/DEPTH_ROADMAP.md index 74343ed..6a87193 100644 --- a/DEPTH_ROADMAP.md +++ b/DEPTH_ROADMAP.md @@ -193,7 +193,10 @@ RBAC 未做，暂以单管理员账号代理；概览口径必须是**系统级* ### [ ] T4.F 健壮性 / 安全 / 性能收口（多为 S，可穿插着做） - [ ] 关键 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 改返 error；rag.DeleteDoc 三库全试+聚合错误(原只回 Neo4j 的错)。 + **live：杀 mcp-go→删→502「未删除请重试」+文档保留；mcp-go 活→删→200+清空**（清场僵尸进程后验证）。 - [ ] 预算硬顶兜底（budget.go ≤0 即无限，单任务可无限烧）| M - [ ] 审批 checkpoint 落盘失败重试（compose_compiler.go:323 现只 log → 任务永卡 waiting）| M - [~] 安全收口（部分）： diff --git a/sundynix-gateway/internal/blob/minio.go b/sundynix-gateway/internal/blob/minio.go index 02b1391..510c434 100644 --- a/sundynix-gateway/internal/blob/minio.go +++ b/sundynix-gateway/internal/blob/minio.go @@ -69,7 +69,7 @@ func (s *Store) Get(ctx context.Context, key string) (string, error) { return string(b), nil } -// Delete 删除对象（best-effort）。 -func (s *Store) Delete(ctx context.Context, key string) { - _ = s.cli.RemoveObject(ctx, s.bucket, key, minio.RemoveObjectOptions{}) +// Delete 删除对象；返回错误供级联删按需失败闭合（其它 best-effort 调用点可忽略返回值）。 +func (s *Store) Delete(ctx context.Context, key string) error { + return s.cli.RemoveObject(ctx, s.bucket, key, minio.RemoveObjectOptions{}) } diff --git a/sundynix-gateway/internal/handler/kb.go b/sundynix-gateway/internal/handler/kb.go index 1f4dbfd..0717159 100644 --- a/sundynix-gateway/internal/handler/kb.go +++ b/sundynix-gateway/internal/handler/kb.go @@ -168,22 +168,30 @@ func (h *Handler) KbDeleteDoc(c *gin.Context) { return } 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"), &contract.ToolCall{Tool: "kb_delete", Args: map[string]any{"kb": scoped, "file_id": d.ID}}); e != nil || res == nil || !res.OK { msg := "kb_delete 无响应" if e != nil { msg = e.Error() - } else if res != nil { + } else if res != nil && 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() { - 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 { c.JSON(http.StatusBadGateway, gin.H{"error": e.Error()}) return diff --git a/sundynix-mcp-go/internal/rag/bleve.go b/sundynix-mcp-go/internal/rag/bleve.go index 11a18f6..b9a660f 100644 --- a/sundynix-mcp-go/internal/rag/bleve.go +++ b/sundynix-mcp-go/internal/rag/bleve.go @@ -101,9 +101,9 @@ func (b *bleveStore) index(kb, doc string, texts []string) error { } // deleteDoc 删除某 (kb, doc) 的全部全文块（笔记重入库前清旧块）。 -func (b *bleveStore) deleteDoc(kb, doc string) { +func (b *bleveStore) deleteDoc(kb, doc string) error { if !b.ready() || doc == "" { - return + return nil } kq := bleve.NewTermQuery(kb) kq.SetField("kb") @@ -113,13 +113,13 @@ func (b *bleveStore) deleteDoc(kb, doc string) { req.Size = 1000 res, err := b.idx.Search(req) if err != nil { - return + return err } batch := b.idx.NewBatch() for _, h := range res.Hits { batch.Delete(h.ID) } - _ = b.idx.Batch(batch) + return b.idx.Batch(batch) } // search 全文检索（可按 kb 过滤），返回 BM25 排序的命中。 diff --git a/sundynix-mcp-go/internal/rag/milvus.go b/sundynix-mcp-go/internal/rag/milvus.go index 129cde2..70ba642 100644 --- a/sundynix-mcp-go/internal/rag/milvus.go +++ b/sundynix-mcp-go/internal/rag/milvus.go @@ -116,14 +116,16 @@ func (m *milvusStore) deleteDoc(ctx context.Context, kb, doc string, dim int) { } // 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 == "" { - return + return nil } expr := fmt.Sprintf("kb == %q && doc == %q", kb, fileID) 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)。 diff --git a/sundynix-mcp-go/internal/rag/rag.go b/sundynix-mcp-go/internal/rag/rag.go index 23f09b5..dc97a71 100644 --- a/sundynix-mcp-go/internal/rag/rag.go +++ b/sundynix-mcp-go/internal/rag/rag.go @@ -5,6 +5,7 @@ package rag import ( "context" "errors" + "fmt" "log" "os" "strconv" @@ -253,16 +254,24 @@ func (e *Engine) Search(ctx context.Context, kb, query string, topK int) ([]Hit, } // 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 { if fileID == "" { return errors.New("file_id 必填") } - e.mv.deleteByFile(ctx, kb, fileID) - e.bleve.deleteDoc(kb, fileID) + var errs []error + 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 { - log.Printf("[rag] 图谱按 file_id 删除失败: %v", err) - return err + errs = append(errs, fmt.Errorf("图谱: %w", err)) + } + if len(errs) > 0 { + return fmt.Errorf("级联删部分失败（幂等，可重试）: %w", errors.Join(errs...)) } log.Printf("[rag] 已删除文档痕迹 kb=%s file_id=%s（向量/全文/图谱）", kb, fileID) return nil