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

[liuhan]

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 <741550...@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
        }