(skywalking-banyandb) 01/01: fix(stream): fix migration corruption from arena tagValue double-free

[li...@apache.org]

This is an automated email from the ASF dual-hosted git repository.

mrproliu pushed a commit to branch stream-migration
in repository https://gitbox.apache.org/repos/asf/skywalking-banyandb.git

commit b54d90d129c1c1704a9146e4aa985da3f034671f
Author: mrproliu <74...@qq.com>
AuthorDate: Mon Jul 6 13:44:34 2026 +0800

    fix(stream): fix migration corruption from arena tagValue double-free
---
 banyand/stream/migration_copy.go | 22 ++++++++++++++++++++--
 1 file changed, 20 insertions(+), 2 deletions(-)

diff --git a/banyand/stream/migration_copy.go b/banyand/stream/migration_copy.go
index ba00988af..82ba59f33 100644
--- a/banyand/stream/migration_copy.go
+++ b/banyand/stream/migration_copy.go
@@ -590,6 +590,24 @@ func releaseStreamSlowCopyArena(a *streamSlowCopyArena) {
 	streamSlowCopyArenaPool.Put(a)
 }
 
+// releaseArenaElements returns an elements to the pool WITHOUT releasing its
+// tagValues into the tagValue pool. The slow-copy path fills elements with
+// tagValues owned by a per-call streamSlowCopyArena; releasing them via the
+// normal releaseElements would put arena-backed *tagValue pointers into the
+// global tagValuePool, where they alias memory the arena still owns and reuses —
+// corrupting later writes. Clearing tagFamilies drops the arena pointers so none
+// linger in the pooled elements.
+func releaseArenaElements(e *elements) {
+	e.seriesIDs = e.seriesIDs[:0]
+	e.timestamps = e.timestamps[:0]
+	e.elementIDs = e.elementIDs[:0]
+	for i := range e.tagFamilies {
+		e.tagFamilies[i] = nil
+	}
+	e.tagFamilies = e.tagFamilies[:0]
+	elementsPool.Put(e)
+}
+
 // appendStreamBlockRowToBuckets routes one row to the correct per-aligned-segment bucket.
 func appendStreamBlockRowToBuckets(
 	ir storage.IntervalRule,
@@ -701,7 +719,7 @@ func slowCopyOneStreamPart(in streamProcessPartInput, srcPartID uint64) (streamP
 	defer releaseLive()
 	defer func() {
 		for _, el := range buckets {
-			releaseElements(el)
+			releaseArenaElements(el)
 		}
 	}()
 
@@ -806,7 +824,7 @@ func runStreamFlushWorker(flushCh <-chan streamFlushJob, fileSystem fs.FileSyste
 }
 
 func directCopyStreamFlushBucket(el *elements, fileSystem fs.FileSystem, partPath string) (sz int64, err error) {
-	defer releaseElements(el)
+	defer releaseArenaElements(el)
 	if mkErr := os.MkdirAll(filepath.Dir(partPath), storage.DirPerm); mkErr != nil {
 		return 0, mkErr
 	}