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 <[email protected]> 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 }
