This is an automated email from the ASF dual-hosted git repository.
hanahmily pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/skywalking-banyandb.git
The following commit(s) were added to refs/heads/main by this push:
new 9b83f0bc6 fix(stream): fix migration corruption from arena tagValue
double-free (#1205)
9b83f0bc6 is described below
commit 9b83f0bc699c5b144e25a979b3d298a677b760b8
Author: mrproliu <[email protected]>
AuthorDate: Mon Jul 6 15:52:21 2026 +0800
fix(stream): fix migration corruption from arena tagValue double-free
(#1205)
---
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..f1c158e48 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 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
}