You are viewing a plain text version of this content. The canonical link for it is here.
Posted to notifications@skywalking.apache.org by ha...@apache.org on 2026/04/30 05:11:27 UTC
(skywalking-banyandb) 23/24: chore(version): Cherry-pick bug fixes from main for v0.10.2
This is an automated email from the ASF dual-hosted git repository.
hanahmily pushed a commit to branch v0.10.x
in repository https://gitbox.apache.org/repos/asf/skywalking-banyandb.git
commit 15259e5f4a1d40cd9da43ea9f2da2a41bc23b407
Author: Hongtao Gao <ha...@gmail.com>
AuthorDate: Wed Apr 29 12:51:26 2026 +0000
chore(version): Cherry-pick bug fixes from main for v0.10.2
Update CHANGES.md for 0.10.2 release and fix cherry-pick conflicts
in segment.go and merger_test.go to align with v0.10.x codebase.
via [HAPI](https://hapi.run)
Co-Authored-By: HAPI <no...@hapi.run>
Co-Authored-By: Claude Opus 4.7 <no...@anthropic.com>
---
CHANGES.md | 33 +++
banyand/internal/storage/segment.go | 72 +++----
banyand/trace/merger_test.go | 417 ------------------------------------
3 files changed, 59 insertions(+), 463 deletions(-)
diff --git a/CHANGES.md b/CHANGES.md
index 56a73bfb3..96ad037ec 100644
--- a/CHANGES.md
+++ b/CHANGES.md
@@ -2,6 +2,39 @@
Release Notes.
+## 0.10.2
+
+### Bug Fixes
+
+- Fix reuse of byte arrays in min/max implementation causing data corruption.
+- Fix index-mode measure queries returning documents outside requested time range.
+- Fix nil pointer panic in segment collectMetrics during shutdown.
+- Fix property schema client connection instability after data node restart.
+- Fix take snapshot error when no data in the segment.
+- Fix(storage): disable rotation task on warm and cold lifecycle nodes.
+- Fix(storage): prevent epoch segment creation from zero timestamps.
+- Fix(sidx): use MinTimestamp/MaxTimestamp instead of SegmentID in streaming sync.
+- Fix(handoff): prevent size limit bypass and sidx timestamp corruption in handoff replay.
+- Fix(handoff): prevent enqueuing parts for online nodes via shared LocateAll.
+- Fix wrong backup path of schema property.
+- Fix OOM issue during migration when a group contains a large amount of data.
+- Fix lifecycle migration failure when the target stage has `close: true`.
+- Fix stale sync request blocking watch session channel.
+- Fix nil pointer panic in disk monitor during early initialization.
+- Fix FileSystemError not matching io/fs.ErrNotExist sentinel.
+- Fix(topn): deduplicate entities in TopN aggregation query.
+- Fix(stream): skip element-index visit when idx/ is absent.
+- Fix(metadata): widen FODC inspection broadcast deadline and parallelize InspectAll.
+- Fix(measure,stream,trace): eliminate flusher/introducer data race on shutdown.
+- Fix: use `topic` instead of `session_id` as the Prometheus label.
+- Fix(fodc): heal reconnect deadlock and add error message when no agents for lifecycle request.
+- Fix(MCP): add explicit validation for properties and tools, and harden the server.
+
+### Chores
+
+- Upgrade Go and npm dependencies for CVE fixes.
+- Bump ui and mcp npm dependencies for CVE fixes.
+
## 0.10.1
### Bug Fixes
diff --git a/banyand/internal/storage/segment.go b/banyand/internal/storage/segment.go
index 04a333507..e4239e0bc 100644
--- a/banyand/internal/storage/segment.go
+++ b/banyand/internal/storage/segment.go
@@ -19,7 +19,6 @@ package storage
import (
"context"
- "encoding/json"
"fmt"
"io/fs"
"path"
@@ -34,19 +33,18 @@ import (
"github.com/apache/skywalking-banyandb/api/common"
commonv1 "github.com/apache/skywalking-banyandb/api/proto/banyandb/common/v1"
+ "github.com/apache/skywalking-banyandb/pkg/convert"
banyanfs "github.com/apache/skywalking-banyandb/pkg/fs"
"github.com/apache/skywalking-banyandb/pkg/index/inverted"
"github.com/apache/skywalking-banyandb/pkg/logger"
"github.com/apache/skywalking-banyandb/pkg/timestamp"
)
-var (
- // ErrSegmentClosed is returned when trying to access a closed segment.
- ErrSegmentClosed = errors.New("segment closed")
- // ErrInvalidSegmentTimestamp is returned when a segment timestamp is zero or near-epoch,
- // indicating a corrupted timestamp (e.g., unset MinTimestamp flowing through sync paths).
- ErrInvalidSegmentTimestamp = errors.New("invalid segment timestamp: epoch or near-epoch time is not valid for APM data")
-)
+// ErrSegmentClosed is returned when trying to access a closed segment.
+var ErrSegmentClosed = errors.New("segment closed")
+
+// ErrInvalidSegmentTimestamp is returned when a segment has an invalid timestamp (e.g., epoch/zero).
+var ErrInvalidSegmentTimestamp = errors.New("invalid segment timestamp: epoch or zero timestamp")
var _ Cache = (*segmentCache)(nil)
@@ -482,61 +480,50 @@ func (sc *segmentController[T, O]) parse(value string) (time.Time, error) {
func (sc *segmentController[T, O]) open() error {
sc.Lock()
defer sc.Unlock()
- invalidSegments := make([]string, 0)
+ emptySegments := make([]string, 0)
err := loadSegments(sc.location, segPathPrefix, sc, sc.getOptions().SegmentInterval, func(start, end time.Time) error {
suffix := sc.format(start)
segmentPath := path.Join(sc.location, fmt.Sprintf(segTemplate, suffix))
// Detect epoch/near-epoch segments created by zero MinTimestamp in sync paths.
// BanyanDB is an APM database -- no legitimate data predates the year 2000.
if start.UnixNano() <= 0 {
- invalidSegments = append(invalidSegments, segmentPath)
+ emptySegments = append(emptySegments, segmentPath)
return nil
}
metadataPath := path.Join(segmentPath, metadataFilename)
- rawMeta, readErr := sc.lfs.Read(metadataPath)
- if readErr != nil {
- if errors.Is(readErr, fs.ErrNotExist) {
- invalidSegments = append(invalidSegments, segmentPath)
+ version, err := sc.lfs.Read(metadataPath)
+ if err != nil {
+ if errors.Is(err, fs.ErrNotExist) {
+ emptySegments = append(emptySegments, segmentPath)
return nil
}
- return readErr
+ return err
}
- if len(rawMeta) == 0 {
- invalidSegments = append(invalidSegments, segmentPath)
+ if len(version) == 0 {
+ emptySegments = append(emptySegments, segmentPath)
return nil
}
- meta, parseErr := readSegmentMeta(rawMeta)
- if parseErr != nil {
- return parseErr
- }
- segmentEnd := end
- if meta.EndTime != "" {
- parsedEnd, timeErr := time.Parse(time.RFC3339Nano, meta.EndTime)
- if timeErr != nil {
- return timeErr
- }
- segmentEnd = parsedEnd
+ if err = checkVersion(convert.BytesToString(version)); err != nil {
+ return err
}
- _, loadErr := sc.load(start, segmentEnd, sc.location)
- return loadErr
+ _, err = sc.load(start, end, sc.location)
+ return err
})
- if len(invalidSegments) > 0 {
- sc.l.Warn().Strs("segments", invalidSegments).Msg("invalid segments found (empty or epoch-dated), removing them")
- for i := range invalidSegments {
- sc.lfs.MustRMAll(invalidSegments[i])
+ if len(emptySegments) > 0 {
+ sc.l.Warn().Strs("segments", emptySegments).Msg("empty segments found, removing them.")
+ for i := range emptySegments {
+ sc.lfs.MustRMAll(emptySegments[i])
}
}
return err
}
func (sc *segmentController[T, O]) create(start time.Time) (*segment[T, O], error) {
- // Reject epoch/near-epoch timestamps caused by zero MinTimestamp flowing through sync paths.
- // BanyanDB is an APM database -- no legitimate data predates the year 2000.
+ sc.Lock()
+ defer sc.Unlock()
if start.UnixNano() <= 0 {
return nil, ErrInvalidSegmentTimestamp
}
- sc.Lock()
- defer sc.Unlock()
last := len(sc.lst) - 1
for i := range sc.lst {
s := sc.lst[last-i]
@@ -564,14 +551,7 @@ func (sc *segmentController[T, O]) create(start time.Time) (*segment[T, O], erro
}
segPath := path.Join(sc.location, fmt.Sprintf(segTemplate, sc.format(start)))
sc.lfs.MkdirPanicIfExist(segPath, DirPerm)
- meta := segmentMeta{
- Version: currentVersion,
- EndTime: end.Format(time.RFC3339Nano),
- }
- data, marshalErr := json.Marshal(meta)
- if marshalErr != nil {
- logger.Panicf("cannot marshal segment metadata: %s", marshalErr)
- }
+ data := []byte(currentVersion)
metadataPath := filepath.Join(segPath, metadataFilename)
lf, err := sc.lfs.CreateLockFile(metadataPath, FilePerm)
if err != nil {
diff --git a/banyand/trace/merger_test.go b/banyand/trace/merger_test.go
index f091cd166..f162b400e 100644
--- a/banyand/trace/merger_test.go
+++ b/banyand/trace/merger_test.go
@@ -1456,420 +1456,3 @@ func Test_flush_cleansUpOnSidxFlushError(t *testing.T) {
sidxPartPath := filepath.Join(tmpPath, sidxDirName, "idx1", fmt.Sprintf("%016x", partID))
require.False(t, fileSystem.IsExist(sidxPartPath), "sidx part directory should be removed on sidx flush failure")
}
-
-func Test_computeSmallMergeThreshold(t *testing.T) {
- threshold := computeSmallMergeThreshold()
- require.GreaterOrEqual(t, threshold, uint64(defaultSmallMergeThreshold),
- "threshold should be at least the default (%d bytes)", defaultSmallMergeThreshold)
-}
-
-func Test_getPartsToMerge_skipsInFlightParts(t *testing.T) {
- tmpPath, defFn := test.Space(require.New(t))
- defer defFn()
-
- fileSystem := fs.NewLocalFileSystem()
-
- // Create 4 file parts
- var allParts []*partWrapper
- for i := range 4 {
- partID := uint64(100 + i)
- mp := generateMemPart()
- mp.mustInitFromTraces(tsTS1)
- mp.mustFlush(fileSystem, partPath(tmpPath, partID))
- p := mustOpenFilePart(partID, tmpPath, fileSystem)
- p.partMetadata.ID = partID
- allParts = append(allParts, newPartWrapper(nil, p))
- releaseMemPart(mp)
- }
-
- closer := run.NewCloser(1)
- defer closer.Done()
-
- tst := &tsTable{
- pm: protector.Nop{},
- fileSystem: fileSystem,
- root: tmpPath,
- loopCloser: closer,
- l: logger.GetLogger("test"),
- curPartID: 200,
- option: option{
- mergePolicy: newDefaultMergePolicyForTesting(),
- },
- }
-
- snp := &snapshot{
- parts: allParts,
- epoch: 1,
- }
- snp.incRef()
- defer snp.decRef()
-
- // Without inFlight: all parts should be eligible
- dst, _ := tst.getPartsToMerge(snp, uint64(1<<30), nil)
- require.GreaterOrEqual(t, len(dst), 2, "should find parts to merge when none are in-flight")
-
- // Mark part 100 and 101 as in-flight
- tst.inFlight = make(map[uint64]struct{})
- tst.inFlight[100] = struct{}{}
- tst.inFlight[101] = struct{}{}
-
- // With inFlight: parts 100 and 101 should be skipped
- dst2, _ := tst.getPartsToMerge(snp, uint64(1<<30), nil)
- if len(dst2) > 0 {
- for _, pw := range dst2 {
- require.NotEqual(t, uint64(100), pw.ID(), "part 100 should be skipped (in-flight)")
- require.NotEqual(t, uint64(101), pw.ID(), "part 101 should be skipped (in-flight)")
- }
- }
-
- // Mark all parts as in-flight: no parts should be eligible
- tst.inFlight[102] = struct{}{}
- tst.inFlight[103] = struct{}{}
- dst3, _ := tst.getPartsToMerge(snp, uint64(1<<30), nil)
- require.Nil(t, dst3, "no parts should be eligible when all are in-flight")
-}
-
-func Test_dispatchAllMerges_laneClassification(t *testing.T) {
- tmpPath, defFn := test.Space(require.New(t))
- defer defFn()
-
- fileSystem := fs.NewLocalFileSystem()
-
- closer := run.NewCloser(1)
-
- tst := &tsTable{
- pm: protector.Nop{},
- fileSystem: fileSystem,
- root: tmpPath,
- loopCloser: closer,
- l: logger.GetLogger("test"),
- curPartID: 200,
- option: option{
- mergePolicy: newDefaultMergePolicyForTesting(),
- },
- }
-
- // Use a very low threshold to force all merges into slow lane
- smallThreshold := uint64(1) // 1 byte - everything is "large"
- slowCh := make(chan *mergeDispatchRequest, 10)
- fastCh := make(chan *mergeDispatchRequest, 10)
-
- // Create 3 file parts and build a snapshot
- var allParts []*partWrapper
- for i := range 3 {
- partID := uint64(200 + i)
- mp := generateMemPart()
- mp.mustInitFromTraces(tsTS1)
- mp.mustFlush(fileSystem, partPath(tmpPath, partID))
- p := mustOpenFilePart(partID, tmpPath, fileSystem)
- p.partMetadata.ID = partID
- allParts = append(allParts, newPartWrapper(nil, p))
- releaseMemPart(mp)
- }
-
- snp := &snapshot{
- parts: allParts,
- epoch: 1,
- }
- snp.incRef()
- tst.snapshot = snp
- defer snp.decRef()
-
- // Drain introductions
- merges := make(chan *mergerIntroduction, 10)
- go func() {
- for m := range merges {
- _ = m
- }
- }()
-
- // Run dispatchAllMerges - it should dispatch to slowCh since threshold is 1 byte
- dispatchDone := make(chan struct{})
- go func() {
- defer close(dispatchDone)
- tst.dispatchAllMerges(smallThreshold, fastCh, slowCh)
- }()
-
- // Should receive on slowCh
- select {
- case req := <-slowCh:
- require.Equal(t, mergeTypeFile, req.typ)
- require.Equal(t, mergeLaneSlow, req.lane)
- require.GreaterOrEqual(t, len(req.parts), 2, "should merge at least 2 parts")
- // Verify inFlight was set
- tst.inFlightMu.RLock()
- for _, pw := range req.parts {
- _, exists := tst.inFlight[pw.ID()]
- require.True(t, exists, "part %d should be in inFlight map", pw.ID())
- }
- tst.inFlightMu.RUnlock()
- // Don't clear inFlight — dispatchAllMerges loop will find no eligible parts and exit naturally
- case <-fastCh:
- t.Fatal("should not dispatch to fast lane with threshold=1")
- case <-time.After(5 * time.Second):
- t.Fatal("timed out waiting for dispatch")
- }
-
- // Wait for dispatchAllMerges to finish (it exits when all parts are in inFlight)
- select {
- case <-dispatchDone:
- case <-time.After(5 * time.Second):
- t.Fatal("dispatchAllMerges did not finish")
- }
-
- // Close the closer after dispatchAllMerges has exited
- closer.Done()
- closer.CloseThenWait()
- close(fastCh)
- close(slowCh)
- close(merges)
-}
-
-func Test_mergeLaneWorker_cleansUpInFlight(t *testing.T) {
- tmpPath, defFn := test.Space(require.New(t))
- defer defFn()
-
- fileSystem := fs.NewLocalFileSystem()
-
- closer := run.NewCloser(1)
-
- tst := &tsTable{
- pm: protector.Nop{},
- fileSystem: fileSystem,
- root: tmpPath,
- loopCloser: closer,
- l: logger.GetLogger("test"),
- curPartID: 0,
- option: option{
- mergePolicy: newDefaultMergePolicyForTesting(),
- },
- inFlight: make(map[uint64]struct{}),
- }
-
- // Create 2 file parts to merge
- var parts []*partWrapper
- for i := range 2 {
- partID := uint64(300 + i)
- mp := generateMemPart()
- mp.mustInitFromTraces(tsTS1)
- mp.mustFlush(fileSystem, partPath(tmpPath, partID))
- p := mustOpenFilePart(partID, tmpPath, fileSystem)
- p.partMetadata.ID = partID
- parts = append(parts, newPartWrapper(nil, p))
- releaseMemPart(mp)
- }
-
- // Mark parts as inFlight
- toBeMerged := make(map[uint64]struct{})
- for _, pw := range parts {
- tst.inFlight[pw.ID()] = struct{}{}
- toBeMerged[pw.ID()] = struct{}{}
- }
-
- require.Len(t, tst.inFlight, 2)
-
- // Create channels
- ch := make(chan *mergeDispatchRequest, 1)
- merges := make(chan *mergerIntroduction, 1)
-
- // Start a mock introducer to consume from merges and close applied.
- mockIntroducerDone := make(chan struct{})
- go func() {
- defer close(mockIntroducerDone)
- for mi := range merges {
- if mi.newPart != nil {
- mi.newPart.decRef()
- }
- if mi.applied != nil {
- close(mi.applied)
- }
- }
- }()
-
- // Start worker in background
- workerDone := make(chan struct{})
- go func() {
- defer close(workerDone)
- tst.mergeLaneWorker(ch, merges)
- }()
-
- // Send a dispatch request
- ch <- &mergeDispatchRequest{
- parts: parts,
- toBeMerged: toBeMerged,
- typ: mergeTypeFile,
- lane: mergeLaneFast,
- }
- close(ch)
-
- // Close the closer to signal shutdown.
- time.Sleep(100 * time.Millisecond)
- closer.Done()
- closer.CloseThenWait()
-
- // Wait for worker to finish
- select {
- case <-workerDone:
- case <-time.After(5 * time.Second):
- t.Fatal("worker did not finish")
- }
- close(merges)
- <-mockIntroducerDone
-
- // Verify inFlight was cleaned up
- tst.inFlightMu.RLock()
- require.Empty(t, tst.inFlight, "inFlight map should be empty after worker completes")
- tst.inFlightMu.RUnlock()
-}
-
-func Test_mergePartsWithConflictingTagTypes(t *testing.T) {
- tests := []struct {
- name string
- parts []*traces
- wantTagTypes map[string]pbv1.ValueType
- wantDeleted []string
- }{
- {
- name: "multiple parts with int and str for same tag",
- parts: []*traces{
- {
- traceIDs: []string{"trace1"},
- timestamps: []int64{1},
- tags: [][]*tagValue{
- {
- {tag: "state", valueType: pbv1.ValueTypeInt64, value: convert.Int64ToBytes(100)},
- {tag: "name", valueType: pbv1.ValueTypeStr, value: []byte("svc1")},
- },
- },
- spans: [][]byte{[]byte("span1")},
- spanIDs: []string{"span1"},
- },
- {
- traceIDs: []string{"trace2"},
- timestamps: []int64{2},
- tags: [][]*tagValue{
- {
- {tag: "state", valueType: pbv1.ValueTypeStr, value: []byte("active")},
- {tag: "name", valueType: pbv1.ValueTypeStr, value: []byte("svc2")},
- },
- },
- spans: [][]byte{[]byte("span2")},
- spanIDs: []string{"span2"},
- },
- },
- wantTagTypes: map[string]pbv1.ValueType{
- "state#int": pbv1.ValueTypeInt64,
- "state#str": pbv1.ValueTypeStr,
- "name": pbv1.ValueTypeStr,
- },
- wantDeleted: []string{"state"},
- },
- {
- name: "merge typed-suffix part with plain part",
- parts: []*traces{
- {
- traceIDs: []string{"trace1"},
- timestamps: []int64{1},
- tags: [][]*tagValue{
- {{tag: "state#int", valueType: pbv1.ValueTypeInt64, value: convert.Int64ToBytes(100)}},
- },
- spans: [][]byte{[]byte("span1")},
- spanIDs: []string{"span1"},
- },
- {
- traceIDs: []string{"trace2"},
- timestamps: []int64{2},
- tags: [][]*tagValue{
- {{tag: "state", valueType: pbv1.ValueTypeStr, value: []byte("pending")}},
- },
- spans: [][]byte{[]byte("span2")},
- spanIDs: []string{"span2"},
- },
- },
- wantTagTypes: map[string]pbv1.ValueType{
- "state#int": pbv1.ValueTypeInt64,
- "state#str": pbv1.ValueTypeStr,
- },
- wantDeleted: []string{"state"},
- },
- }
-
- for _, tt := range tests {
- t.Run(tt.name, func(t *testing.T) {
- tmpPath, defFn := test.Space(require.New(t))
- defer defFn()
- fileSystem := fs.NewLocalFileSystem()
-
- partIDs := make([]uint64, len(tt.parts))
- for idx := range partIDs {
- partIDs[idx] = uint64(idx)
- }
- dstID := uint64(999)
- dstPath := mergePartsWithConflictTags(t, fileSystem, tmpPath, tt.parts, partIDs, dstID)
-
- mergedPart := mustOpenFilePart(dstID, tmpPath, fileSystem)
- for tag, wantType := range tt.wantTagTypes {
- gotType, exists := mergedPart.tagType[tag]
- require.True(t, exists, "expected tag type entry for %q", tag)
- require.Equal(t, wantType, gotType, "tag type mismatch for %q", tag)
- }
- for _, tag := range tt.wantDeleted {
- _, ok := mergedPart.tagType[tag]
- require.False(t, ok, "tag %q should not exist in tagType after merge", tag)
- dataFile := filepath.Join(dstPath, tag+tagsFilenameExt)
- require.False(t, fileSystem.IsExist(dataFile), "tag data file %q should not exist", dataFile)
- metaFile := filepath.Join(dstPath, tag+tagsMetadataFilenameExt)
- require.False(t, fileSystem.IsExist(metaFile), "tag metadata file %q should not exist", metaFile)
- }
- })
- }
-}
-
-func mergePartsWithConflictTags(t *testing.T, fileSystem fs.FileSystem, tmpPath string, tsList []*traces, partIDs []uint64, dstID uint64) string {
- t.Helper()
- var parts []*partWrapper
- for i, traces := range tsList {
- mp := generateMemPart()
- mp.mustInitFromTraces(traces)
- mp.mustFlush(fileSystem, partPath(tmpPath, partIDs[i]))
- p := mustOpenFilePart(partIDs[i], tmpPath, fileSystem)
- parts = append(parts, &partWrapper{p: p})
- releaseMemPart(mp)
- }
-
- conflictTags := collectConflictTags(parts)
-
- var pmi []*partMergeIter
- var traceSize uint64
- for i := range tsList {
- p := mustOpenFilePart(partIDs[i], tmpPath, fileSystem)
- iter := generatePartMergeIter()
- iter.mustInitFromPart(p)
- pmi = append(pmi, iter)
- traceSize += uint64(len(tsList[i].traceIDs))
- }
-
- br := generateBlockReader()
- br.init(pmi)
- bw := generateBlockWriter()
- dstPath := partPath(tmpPath, dstID)
- bw.mustInitForFilePart(fileSystem, dstPath, false, int(traceSize))
-
- closeCh := make(chan struct{})
- pm, tf, tagTypes, mergeErr := mergeBlocks(closeCh, bw, br, conflictTags)
- close(closeCh)
- require.NoError(t, mergeErr)
- require.NotNil(t, pm)
-
- releaseBlockWriter(bw)
- releaseBlockReader(br)
- for _, iter := range pmi {
- releasePartMergeIter(iter)
- }
-
- pm.mustWriteMetadata(fileSystem, dstPath)
- tf.mustWriteTraceIDFilter(fileSystem, dstPath)
- tagTypes.mustWriteTagType(fileSystem, dstPath)
- fileSystem.SyncPath(dstPath)
- return dstPath
-}