fix(core): 行写入失败回收新落 blob 链 — 堵住两条链页泄漏路径

- serializeRowInternal: blob 列先行落链, 后续列编码/布局失败时本次新落的
  链成为孤儿页; 现在 defer 兜底只还 owned 链 (本次新落), 透传引用不碰
- InsertRowWithID/InsertRowsWithIDs: 唯一冲突的预检先于任何索引写入,
  冲突时序列化已落的链整行/整批归还; 其它部分写入错误仍靠事务回滚
- blobFieldRef/blobRefFromValue 返回 owned 标记; blobCommit 注释同步
- 回归: DB/core/blob_leak_test.go (编码失败 / 唯一冲突两例)

DB/core/blob_commit.go

@@ -66,8 +66,10 @@ func (db *DB) BlobCommit(r io.Reader, compress types.FieldCompress) (BlobRef, er
 // types/blob.go 里"引擎自动走链页"的设计目标背离.
 //
 // 与 tx.BlobCommit 的唯一差别在失败回收: 有 currentTxn 时新页登记进 tx.allocated, 回滚即回收;
-// 无事务时没有 allocated 可回滚 —— 函数返回 err 后, 已分配的页只能由调用方保证不再复用
-// (core 的列内联写入在出错时整行写入中止, 这批页成为孤儿, 与无事务下任何写入失败的语义一致).
+// 无事务时没有 allocated 可回滚 —— 函数返回 err 后, 已分配的页按"调用方中止写入"语义处理:
+// 调用方是 core 行写入 (serializeRowInternal) 时, 整行写入失败会自动把本次新落的链还掉
+// (见那里的 newBlobRefs 兜底); 调用方直接调 BlobCommit 的 (如 ORM 流式写), 仍需自管
+// 生命周期 —— 成功拿 ref 后归调用方, 流中断返 err 时这批页只能等上层处置。
 func (db *DB) blobCommit(r io.Reader, compress types.FieldCompress) (BlobRef, error) {
 	if db == nil {
 		return BlobRef{}, errors.New("BlobCommit: db is nil")

DB/core/blob_leak_test.go

@@ -0,0 +1,187 @@
+package core_test
+
+// 回归: 写入失败的行不得泄漏 blob 链页 (Type=9 孤儿页)。
+//
+// 两条历史泄漏路径:
+//  1. serializeRowInternal 里 blob 列先落链, 后续列编码失败 → 链页无人引用;
+//  2. indexMeta.InsertRow/InsertRows 唯一冲突报错 → 行没落任何索引, 链页无人引用。
+// 修复后这些链页必须回到空闲链 (orphancheck 的 blob 感知口径孤儿数 == 0)。
+
+import (
+	"bytes"
+	"testing"
+
+	"git.ycl.pub/yl32940/TsDB2/DB/core"
+	"git.ycl.pub/yl32940/TsDB2/DB/core/page"
+	"git.ycl.pub/yl32940/TsDB2/DB/core/types"
+	"git.ycl.pub/yl32940/TsDB2/internal/orphancheck"
+)
+
+func blobLeakSchema() []types.FieldInfo {
+	return []types.FieldInfo{
+		{FieldType: types.FieldType_Uint64, FieldName: []byte("id")},
+		{FieldType: types.FieldType_Blob, FieldName: []byte("body")},
+		{FieldType: types.FieldType_String, Flag: types.WidthCodeFor(32), FieldName: []byte("tag")},
+	}
+}
+
+func openBlobLeakDB(t *testing.T, path string) *core.DB {
+	t.Helper()
+	db, err := core.Open(path)
+	if err != nil {
+		t.Fatal(err)
+	}
+	db.SetAutoCommitDelay(0)
+	tbl, err := db.CreateTable("docs", blobLeakSchema())
+	if err != nil {
+		t.Fatal(err)
+	}
+	if _, err := tbl.CreateIndex("pk_docs", page.IndexTypePrimary, []byte{0}); err != nil {
+		t.Fatal(err)
+	}
+	return db
+}
+
+// 失败行 (序列化阶段: blob 列之后的字符串列超宽) 不得泄漏链页。
+func TestBlobChainFreedOnSerializeError(t *testing.T) {
+	path := t.TempDir() + "/leak1.db"
+	db := openBlobLeakDB(t, path)
+
+	// 先行一行成功数据, 作为"活链"对照。
+	good, ok := db.FindTable("docs")
+	if !ok {
+		t.Fatal("FindTable docs: not found")
+	}
+	if _, err := good.InsertRowWithID(1, []types.FieldValue{
+		{FieldName: []byte("id"), Value: uint64(1)},
+		{FieldName: []byte("body"), Value: bytes.Repeat([]byte("a"), 40000)},
+		{FieldName: []byte("tag"), Value: []byte("ok")},
+	}); err != nil {
+		t.Fatal(err)
+	}
+
+	// 失败行: blob 列先落链 (40KB), 随后 tag 超宽 → 整行写入中止。
+	if _, err := good.InsertRowWithID(2, []types.FieldValue{
+		{FieldName: []byte("id"), Value: uint64(2)},
+		{FieldName: []byte("body"), Value: bytes.Repeat([]byte("b"), 40000)},
+		{FieldName: []byte("tag"), Value: bytes.Repeat([]byte("x"), 100)},
+	}); err == nil {
+		t.Fatal("超宽 tag 应报错, 实际写入成功")
+	}
+
+	if err := db.Commit(); err != nil {
+		t.Fatal(err)
+	}
+	if err := db.Close(); err != nil {
+		t.Fatal(err)
+	}
+
+	R, err := orphancheck.Run(path, orphancheck.Options{EngineBlobScan: true})
+	if err != nil {
+		t.Fatal(err)
+	}
+	if len(R.Orphans) > 0 {
+		t.Fatalf("失败行泄漏 %d 页: %v (按类型 %v)", len(R.Orphans), idsOf(R.Orphans), R.ByType)
+	}
+	if R.BlobPages == 0 {
+		t.Fatal("活 blob 链页数 = 0 —— 成功行的链也应被认领, 检查走查逻辑")
+	}
+}
+
+// 唯一冲突行 (索引阶段预检报错, 零写入) 不得泄漏链页。
+func TestBlobChainFreedOnUniqueConflict(t *testing.T) {
+	path := t.TempDir() + "/leak2.db"
+	db := openBlobLeakDB(t, path)
+	tbl, ok := db.FindTable("docs")
+	if !ok {
+		t.Fatal("FindTable docs: not found")
+	}
+	if _, err := tbl.CreateIndex("uq_docs_tag", page.IndexTypeBTree, []byte{2}, true); err != nil {
+		t.Fatal(err)
+	}
+
+	body1 := bytes.Repeat([]byte("a"), 40000)
+	if _, err := tbl.InsertRowWithID(1, []types.FieldValue{
+		{FieldName: []byte("id"), Value: uint64(1)},
+		{FieldName: []byte("body"), Value: body1},
+		{FieldName: []byte("tag"), Value: []byte("dup")},
+	}); err != nil {
+		t.Fatal(err)
+	}
+	if err := db.Commit(); err != nil {
+		t.Fatal(err)
+	}
+
+	// 冲突行: 唯一预检先于任何写入, blob 链已落 → 必须被回收。
+	if _, err := tbl.InsertRowWithID(2, []types.FieldValue{
+		{FieldName: []byte("id"), Value: uint64(2)},
+		{FieldName: []byte("body"), Value: bytes.Repeat([]byte("b"), 40000)},
+		{FieldName: []byte("tag"), Value: []byte("dup")},
+	}); err == nil {
+		t.Fatal("唯一冲突应报错, 实际写入成功")
+	}
+
+	// 批量路径同样不得泄漏。
+	if _, err := tbl.InsertRowsWithIDs([]uint64{3}, [][]types.FieldValue{{
+		{FieldName: []byte("id"), Value: uint64(3)},
+		{FieldName: []byte("body"), Value: bytes.Repeat([]byte("c"), 40000)},
+		{FieldName: []byte("tag"), Value: []byte("dup")},
+	}}); err == nil {
+		t.Fatal("批量唯一冲突应报错, 实际写入成功")
+	}
+
+	if err := db.Commit(); err != nil {
+		t.Fatal(err)
+	}
+
+	// 先行行的 blob 必须完好 (回收不得误伤活链)。
+	got, err := tbl.BlobRead(types.BlobRef{
+		FirstPageID: firstPageOf(t, tbl, 1), TotalLen: uint32(len(body1)),
+	}, types.CompressNone)
+	if err != nil {
+		t.Fatalf("读回先行行的 blob: %v", err)
+	}
+	if !bytes.Equal(got, body1) {
+		t.Fatalf("先行行的 blob 内容被损毁: 读回 %d 字节, 期望 %d", len(got), len(body1))
+	}
+
+	if err := db.Close(); err != nil {
+		t.Fatal(err)
+	}
+
+	R, err := orphancheck.Run(path, orphancheck.Options{EngineBlobScan: true})
+	if err != nil {
+		t.Fatal(err)
+	}
+	if len(R.Orphans) > 0 {
+		t.Fatalf("唯一冲突行泄漏 %d 页: %v (按类型 %v)", len(R.Orphans), idsOf(R.Orphans), R.ByType)
+	}
+}
+
+// firstPageOf 取一行 body 列的链首页 (经行存储读回 BlobRef)。
+func firstPageOf(t *testing.T, tbl *core.Table, pkID uint64) uint32 {
+	t.Helper()
+	vals, err := tbl.FindByPK(pkID)
+	if err != nil {
+		t.Fatalf("FindByPK(%d): %v", pkID, err)
+	}
+	for _, fv := range vals {
+		if string(fv.FieldName) == "body" {
+			ref, ok := fv.Value.(types.BlobRef)
+			if !ok {
+				t.Fatalf("body 列类型 %T, 期望 types.BlobRef", fv.Value)
+			}
+			return ref.FirstPageID
+		}
+	}
+	t.Fatal("行里没有 body 列")
+	return 0
+}
+
+func idsOf(ps []orphancheck.PageInfo) []uint32 {
+	out := make([]uint32, len(ps))
+	for i, p := range ps {
+		out[i] = p.ID
+	}
+	return out
+}

DB/core/table_crud.go

@@ -1,6 +1,7 @@
 package core
 
 import (
+	"errors"
 	"fmt"
 
 	"git.ycl.pub/yl32940/TsDB2/DB/core/page"
@@ -67,6 +68,12 @@ func (table *Table) InsertRowWithID(pkID uint64, values []types.FieldValue) (uin
 		return 0, err
 	}
 	if err := table.indexMeta.InsertRow(table.Field, pkID, rowData, table.activeVersion, table.versions); err != nil {
+		// 唯一冲突的预检先于任何索引写入 (index_meta_page.InsertRow): 此时行没落进
+		// 任何索引, 序列化时已提交的 blob 链成为孤儿, 必须还掉。其它错误 (页满 / IO)
+		// 可能已部分写入, 不能动链 —— 那种半成品只能靠事务回滚兜底。
+		if errors.Is(err, ErrDuplicatedKey) {
+			_ = table.releaseRowBlobRefs(rowData)
+		}
 		return 0, err
 	}
 	table.info.TotalRows++
@@ -90,8 +97,10 @@ func (table *Table) InsertRowWithID(pkID uint64, values []types.FieldValue) (uin
 //     本身已经近似批量; saveTableInfo / onWrite 摊到批末一次。
 //
 // 失败语义: 序列化阶段报错时, 先前行写入的堆块成为无主垃圾 (泄漏而非错乱,
-// 行数据并未落任何索引, 不会被读到); 索引阶段报错时部分索引已写入 —— 与
-// 单行路径"写了一半报错"一致, 靠事务回滚兜底。
+// 行数据并未落任何索引, 不会被读到); 各先行行的 blob 链在报错时会被 serializeRow
+// 的兜底回收 (见 serializeRowInternal 的 newBlobRefs), 不随堆块一起泄漏。
+// 索引阶段报错时部分索引已写入 —— 与单行路径"写了一半报错"一致, 唯一冲突
+// (预检先于任何写入) 时整批 blob 链会回收, 其余错误靠事务回滚兜底。
 func (table *Table) InsertRowsWithIDs(pkIDs []uint64, values [][]types.FieldValue) ([]uint64, error) {
 	if err := table.errIfReadOnly(); err != nil {
 		return nil, err
@@ -184,6 +193,14 @@ func (table *Table) insertRows(pkIDs []uint64, values [][]types.FieldValue, allR
 
 	// 6) 索引层批量写入
 	if err := table.indexMeta.InsertRows(table.Field, pkIDs, rowDatas, table.activeVersion, table.versions); err != nil {
+		// 唯一预检是"全部查完才动第一笔写入" (index_meta_page.InsertRows):
+		// 唯一冲突意味着零写入, 每行序列化时已落的 blob 链整批还掉。
+		// 非唯一错误可能已部分写入, 链不能动 (同单行路径)。
+		if errors.Is(err, ErrDuplicatedKey) {
+			for _, rd := range rowDatas {
+				_ = table.releaseRowBlobRefs(rd)
+			}
+		}
 		return nil, err
 	}
 

DB/core/table_rowcodec.go

@@ -107,71 +107,74 @@ func (table *Table) dynamicFieldData(masterIdx int, f types.FieldInfo, value int
 //
 // checkOnly 是唯一性试算行 (serializeRowForCheck): 绝不能落盘, 返回零引用即可 ——
 // Blob 列不能当索引键 (见 CreateIndex), 试算行的键字节与它无关。
-func (table *Table) blobFieldRef(f types.FieldInfo, value interface{}, checkOnly bool) (BlobRef, error) {
+func (table *Table) blobFieldRef(f types.FieldInfo, value interface{}, checkOnly bool) (BlobRef, bool, error) {
 	switch v := value.(type) {
 	case BlobRef: // == types.BlobRef (类型别名), 引用原样透传
-		return v, nil
+		return v, false, nil
 	case types.Blob:
 		return table.blobRefFromValue(f, v, checkOnly)
 	case *types.Blob:
 		if v == nil {
-			return BlobRef{}, nil
+			return BlobRef{}, false, nil
 		}
 		return table.blobRefFromValue(f, *v, checkOnly)
 	}
 
 	if checkOnly {
-		return BlobRef{}, nil
+		return BlobRef{}, false, nil
 	}
 
 	var src io.Reader
 	switch v := value.(type) {
 	case []byte:
 		if len(v) == 0 {
-			return BlobRef{}, nil
+			return BlobRef{}, false, nil
 		}
 		src = bytes.NewReader(v)
 	case string:
 		if v == "" {
-			return BlobRef{}, nil
+			return BlobRef{}, false, nil
 		}
 		src = strings.NewReader(v)
 	case io.Reader:
 		src = v
 	default:
-		return BlobRef{}, fmt.Errorf("字段 %q (%s): blob 列只接受 []byte / string / io.Reader / types.Blob / types.BlobRef, 收到 %T",
+		return BlobRef{}, false, fmt.Errorf("字段 %q (%s): blob 列只接受 []byte / string / io.Reader / types.Blob / types.BlobRef, 收到 %T",
 			f.FieldName, types.TypeNameOf(f.FieldType), value)
 	}
 	ref, err := table.db.blobCommit(src, types.CompressNone)
 	if err != nil {
-		return BlobRef{}, fmt.Errorf("字段 %q (%s): blob 落链失败: %w",
+		return BlobRef{}, false, fmt.Errorf("字段 %q (%s): blob 落链失败: %w",
 			f.FieldName, types.TypeNameOf(f.FieldType), err)
 	}
-	return ref, nil
+	return ref, true, nil
 }
 
 // blobRefFromValue 处理 types.Blob 载体: 既有"内容"形态 (Source / Bytes),
 // 也有"读回镜像"形态 (FirstPageID / TotalLen 已由引擎填好).
-func (table *Table) blobRefFromValue(f types.FieldInfo, b types.Blob, checkOnly bool) (BlobRef, error) {
+//
+// 返回值 owned 标注该引用是否本次新落的链: 调用方在【整行写入失败】时只能还
+// 自己新落的链 —— 透传引用 (调用方持有的) 还掉会把别的行/别的表的链误回收。
+func (table *Table) blobRefFromValue(f types.FieldInfo, b types.Blob, checkOnly bool) (BlobRef, bool, error) {
 	hasRef := b.FirstPageID != 0 || b.TotalLen != 0
 	hasContent := b.Source != nil || len(b.Bytes) > 0
 	if hasRef {
 		if hasContent {
-			return BlobRef{}, fmt.Errorf("字段 %q: types.Blob 同时带了引用与内容 (Source/Bytes), 无法判定该用哪个", f.FieldName)
+			return BlobRef{}, false, fmt.Errorf("字段 %q: types.Blob 同时带了引用与内容 (Source/Bytes), 无法判定该用哪个", f.FieldName)
 		}
-		return b.Ref(), nil // 读回镜像: 原样透传
+		return b.Ref(), false, nil // 读回镜像: 原样透传
 	}
 	if !hasContent {
-		return BlobRef{}, nil // 空 blob: 不分配链页
+		return BlobRef{}, false, nil // 空 blob: 不分配链页
 	}
 	if checkOnly {
-		return BlobRef{}, nil
+		return BlobRef{}, false, nil
 	}
 	if b.Source != nil && len(b.Bytes) > 0 {
-		return BlobRef{}, fmt.Errorf("字段 %q: types.Blob 的 Source 与 Bytes 都非空, 引擎无法选择", f.FieldName)
+		return BlobRef{}, false, fmt.Errorf("字段 %q: types.Blob 的 Source 与 Bytes 都非空, 引擎无法选择", f.FieldName)
 	}
 	if !b.Compress.IsValid() {
-		return BlobRef{}, fmt.Errorf("字段 %q: types.Blob.Compress 非法", f.FieldName)
+		return BlobRef{}, false, fmt.Errorf("字段 %q: types.Blob.Compress 非法", f.FieldName)
 	}
 	var src io.Reader
 	if b.Source != nil {
@@ -181,10 +184,10 @@ func (table *Table) blobRefFromValue(f types.FieldInfo, b types.Blob, checkOnly
 	}
 	ref, err := table.db.blobCommit(src, b.Compress)
 	if err != nil {
-		return BlobRef{}, fmt.Errorf("字段 %q (%s): blob 落链失败: %w",
+		return BlobRef{}, false, fmt.Errorf("字段 %q (%s): blob 落链失败: %w",
 			f.FieldName, types.TypeNameOf(f.FieldType), err)
 	}
-	return ref, nil
+	return ref, true, nil
 }
 
 // serializeRow 将 FieldValue 切片序列化为行数据字节 (不含 pkID 前缀)
@@ -222,7 +225,22 @@ func nullFieldError(name string, ft types.FieldType) error {
 		types.ErrValueRange, name, types.TypeNameOf(ft))
 }
 
-func (table *Table) serializeRowInternal(values []types.FieldValue, oldRowData []byte, checkOnly bool) ([]byte, error) {
+func (table *Table) serializeRowInternal(values []types.FieldValue, oldRowData []byte, checkOnly bool) (_ []byte, err error) {
+	// 失败回收: blob 列在 Pass 1 就落链 (行内只有 8B 引用), 一旦后续列编码失败 /
+	// 布局自检失败, 这批新链页还不掉就成为孤儿页 (无事务写入的既有泄漏路径,
+	// 见 blobCommit 的注释)。只还"本次新落的链" (owned) —— 透传引用
+	// (调用方持有的 types.BlobRef / 读回镜像) 不属于本行, 绝不能碰。
+	// 这些链此刻尚未被任何行/索引引用, 无条件归还安全; 更新路径 Pass 3 已释放的
+	// 旧链不会出现在这里 (owned 只记新链)。
+	var newBlobRefs []BlobRef
+	defer func() {
+		if err != nil && !checkOnly {
+			for _, r := range newBlobRefs {
+				_ = table.BlobFree(r)
+			}
+		}
+	}()
+
 	ver := table.pickVersionForUpdate(oldRowData, values)
 
 	// 写行闸门 (第三处行宽校验)。
@@ -391,10 +409,13 @@ func (table *Table) serializeRowInternal(values []types.FieldValue, oldRowData [
 			// Blob 列行内只有 8B BlobRef, 但用户手上多半是"内容本身" ([]byte / io.Reader),
 			// 由引擎在这里把内容落进链页、换成引用 (见 blobFieldRef). 已经是引用的原样用。
 			// 试算行 (checkOnly) 不落盘: Blob 列不能当索引键 (见 CreateIndex), 键字节与它无关。
-			ref, err := table.blobFieldRef(f, value, checkOnly)
+			ref, owned, err := table.blobFieldRef(f, value, checkOnly)
 			if err != nil {
 				return nil, err
 			}
+			if owned {
+				newBlobRefs = append(newBlobRefs, ref)
+			}
 			enc, err := table.encodeFieldValue(int(masterIdx), f, ref)
 			if err != nil {
 				return nil, err