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/08/08 23:51:59 UTC
(skywalking-banyandb) branch trace-merge-resilience created (now 20c2a2edd)
This is an automated email from the ASF dual-hosted git repository.
hanahmily pushed a change to branch trace-merge-resilience
in repository https://gitbox.apache.org/repos/asf/skywalking-banyandb.git
at 20c2a2edd docs: record the trace merge failure resilience fixes in CHANGES
This branch includes the following new commits:
new 241240593 fix(storage): remove partially-written merge output on failure
new cc8b6f922 feat(trace): quarantine unmergeable parts from merge selection
new 84c6aba87 feat(trace): back off merge dispatch after consecutive failures
new c8f4f4c4e fix(trace): survive panics in the merge workers and dispatcher
new 32435b315 chore(trace): satisfy fieldalignment in the merge backoff code
new 20c2a2edd docs: record the trace merge failure resilience fixes in CHANGES
The 6 revisions listed above as "new" are entirely new to this
repository and will be described in separate emails. The revisions
listed as "add" were already present in the repository and have only
been added to this reference.
(skywalking-banyandb) 05/06: chore(trace): satisfy fieldalignment in the merge backoff code
Posted by ha...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
hanahmily pushed a commit to branch trace-merge-resilience
in repository https://gitbox.apache.org/repos/asf/skywalking-banyandb.git
commit 32435b3159ec6c04981be4ce9fffdd1eb6ed7c90
Author: owenwillison-vajra <30...@qq.com>
AuthorDate: Sat Aug 8 23:35:08 2026 +0000
chore(trace): satisfy fieldalignment in the merge backoff code
Group backoffUntil with the other pointer-bearing fields in
mergeLoopControl so govet fieldalignment is satisfied, and use min for
the backoff cap.
---
banyand/trace/merge_control.go | 7 ++-----
1 file changed, 2 insertions(+), 5 deletions(-)
diff --git a/banyand/trace/merge_control.go b/banyand/trace/merge_control.go
index ced1c6a85..4309874cb 100644
--- a/banyand/trace/merge_control.go
+++ b/banyand/trace/merge_control.go
@@ -40,12 +40,12 @@ type mergeLoopControl struct {
changed chan struct{}
trigger chan struct{}
nowFunc func() time.Time
+ backoffUntil time.Time
version uint64
workVersion uint64
emptyEpoch uint64
emptyWorkVersion uint64
waveMaxPartID uint64
- backoffUntil time.Time
queued int
running int
consecutiveFailures int
@@ -113,10 +113,7 @@ func (mc *mergeLoopControl) recordOutcome(success bool) {
mc.consecutiveFailures++
delay := mergeBackoffCap
if mc.consecutiveFailures <= mergeBackoffMaxShift {
- delay = mergeBackoffBase << uint(mc.consecutiveFailures-1)
- if delay > mergeBackoffCap {
- delay = mergeBackoffCap
- }
+ delay = min(mergeBackoffBase<<uint(mc.consecutiveFailures-1), mergeBackoffCap)
}
mc.backoffUntil = mc.now().Add(delay)
mc.notifyLocked()
(skywalking-banyandb) 02/06: feat(trace): quarantine unmergeable parts from merge selection
Posted by ha...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
hanahmily pushed a commit to branch trace-merge-resilience
in repository https://gitbox.apache.org/repos/asf/skywalking-banyandb.git
commit cc8b6f922130bf7b2cee49482b03f206fbb82a03
Author: owenwillison-vajra <30...@qq.com>
AuthorDate: Sat Aug 8 13:26:15 2026 +0000
feat(trace): quarantine unmergeable parts from merge selection
Wrap merge-read failures in a typed unreadablePartError carrying the
failing part's identity, count consecutive attributable failures per
part, and after three failures exclude the part from both the hot merge
selector and the finalize-round selector. The part stays in the
snapshot (queryable, TTL-deleted as usual); the registry is in-memory
and swept when parts leave the snapshot. New metrics:
total_merge_part_quarantined (counter) and merge_quarantined_parts
(gauge).
---
banyand/trace/finalizer.go | 3 +
banyand/trace/merger.go | 17 ++++
banyand/trace/metrics.go | 35 +++++++-
banyand/trace/part_iter.go | 20 ++++-
banyand/trace/quarantine.go | 85 ++++++++++++++++++
banyand/trace/quarantine_test.go | 184 +++++++++++++++++++++++++++++++++++++++
banyand/trace/tstable.go | 4 +
banyand/trace/unreadable_part.go | 35 ++++++++
8 files changed, 377 insertions(+), 6 deletions(-)
diff --git a/banyand/trace/finalizer.go b/banyand/trace/finalizer.go
index de6910eb2..510f12841 100644
--- a/banyand/trace/finalizer.go
+++ b/banyand/trace/finalizer.go
@@ -102,6 +102,9 @@ func (tst *tsTable) runFinalizeRound(samplers []sdk.Sampler, graceNs int64) (boo
if _, inFlight := tst.inFlight[pw.ID()]; inFlight {
continue
}
+ if tst.isPartQuarantined(pw.ID()) {
+ continue
+ }
if pw.p.partMetadata.FinalizeGen >= gNext {
continue
}
diff --git a/banyand/trace/merger.go b/banyand/trace/merger.go
index ae71470c8..9c81522eb 100644
--- a/banyand/trace/merger.go
+++ b/banyand/trace/merger.go
@@ -489,6 +489,7 @@ func (tst *tsTable) mergeLaneWorker(ch chan *mergeDispatchRequest, merges chan *
if !errors.Is(mergeErr, errClosed) {
tst.l.Logger.Warn().Err(mergeErr).Str("typ", req.typ).Str("lane", req.lane).Msg("merge lane worker error")
tst.incTotalMergeLoopErr(1)
+ tst.recordUnreadablePart(mergeErr)
}
}
}
@@ -957,22 +958,38 @@ func (tst *tsTable) getPartsToMergeUpTo(snapshot *snapshot, freeDiskSize uint64,
maxPartID uint64,
) ([]*partWrapper, map[uint64]struct{}) {
var parts []*partWrapper
+ // Avoid allocating liveIDs on the hot path when nothing is quarantined.
+ sweep := tst.hasQuarantinedParts()
+ var liveIDs map[uint64]struct{}
+ if sweep {
+ liveIDs = make(map[uint64]struct{}, len(snapshot.parts))
+ }
tst.inFlightMu.RLock()
for _, pw := range snapshot.parts {
if pw.mp != nil || pw.p.partMetadata.TotalCount < 1 {
continue
}
+ if sweep {
+ liveIDs[pw.ID()] = struct{}{}
+ }
if maxPartID > 0 && pw.ID() > maxPartID {
continue
}
if _, inFlight := tst.inFlight[pw.ID()]; inFlight {
continue
}
+ if tst.isPartQuarantined(pw.ID()) {
+ continue
+ }
parts = append(parts, pw)
}
tst.inFlightMu.RUnlock()
+ if sweep {
+ tst.sweepQuarantine(liveIDs)
+ }
+
dst = tst.option.mergePolicy.getPartsToMerge(dst, parts, freeDiskSize)
if len(dst) == 0 {
return nil, nil
diff --git a/banyand/trace/metrics.go b/banyand/trace/metrics.go
index dd6846db7..0a7157ece 100644
--- a/banyand/trace/metrics.go
+++ b/banyand/trace/metrics.go
@@ -62,10 +62,11 @@ type metrics struct {
totalFlushIntroLatency meter.Counter
totalFlushLatency meter.Counter
- totalMergedParts meter.Counter
- totalMergeLatency meter.Counter
- totalMerged meter.Counter
- totalMergeQueueLatency meter.Counter
+ totalMergedParts meter.Counter
+ totalMergeLatency meter.Counter
+ totalMerged meter.Counter
+ totalMergeQueueLatency meter.Counter
+ totalMergePartQuarantined meter.Counter
pipelineTracesEvaluated meter.Counter
pipelineTracesDropped meter.Counter
@@ -274,6 +275,14 @@ func (tst *tsTable) incTotalMergeQueueLatency(delta float64, typ, lane string) {
tst.metrics.totalMergeQueueLatency.Inc(delta, typ, lane)
}
+// incTotalMergePartQuarantined counts a part newly crossing the quarantine threshold.
+func (tst *tsTable) incTotalMergePartQuarantined(delta int) {
+ if tst == nil || tst.metrics == nil {
+ return
+ }
+ tst.metrics.totalMergePartQuarantined.Inc(float64(delta))
+}
+
// The pipeline metric increment helpers below are wired into the merge filter by
// the config-driven activation story.
@@ -442,6 +451,7 @@ func (m *metrics) DeleteAll() {
m.totalMerged.Delete("file", "slow")
m.totalMergeQueueLatency.Delete("file", "fast")
m.totalMergeQueueLatency.Delete("file", "slow")
+ m.totalMergePartQuarantined.Delete()
m.pipelineTracesEvaluated.Delete()
m.pipelineTracesDropped.Delete()
@@ -488,6 +498,7 @@ func (s *supplier) newMetrics(p common.Position) storage.Metrics {
totalMergeLatency: factory.NewCounter("total_merge_latency", "type", "lane"),
totalMerged: factory.NewCounter("total_merged", "type", "lane"),
totalMergeQueueLatency: factory.NewCounter("total_merge_queue_latency", "type", "lane"),
+ totalMergePartQuarantined: factory.NewCounter("total_merge_part_quarantined"),
pipelineTracesEvaluated: factory.NewCounter("pipeline_traces_evaluated"),
pipelineTracesDropped: factory.NewCounter("pipeline_traces_dropped"),
pipelineTracesRetained: factory.NewCounter("pipeline_traces_retained"),
@@ -514,6 +525,7 @@ func (s *supplier) newMetrics(p common.Position) storage.Metrics {
totalFilePartBytes: factory.NewGauge("total_file_part_bytes", common.ShardLabelNames()...),
totalFilePartUncompressedBytes: factory.NewGauge("total_file_part_uncompressed_bytes", common.ShardLabelNames()...),
pendingDataCount: factory.NewGauge("pending_data_count", common.ShardLabelNames()...),
+ mergeQuarantinedParts: factory.NewGauge("merge_quarantined_parts", common.ShardLabelNames()...),
},
indexMetrics: inverted.NewMetrics(factory, common.SegLabelNames()...),
}
@@ -549,6 +561,7 @@ func (qs *queueSupplier) newMetrics(p common.Position) (storage.Metrics, observa
totalMergeLatency: factory.NewCounter("total_merge_latency", "type", "lane"),
totalMerged: factory.NewCounter("total_merged", "type", "lane"),
totalMergeQueueLatency: factory.NewCounter("total_merge_queue_latency", "type", "lane"),
+ totalMergePartQuarantined: factory.NewCounter("total_merge_part_quarantined"),
pipelineTracesEvaluated: factory.NewCounter("pipeline_traces_evaluated"),
pipelineTracesDropped: factory.NewCounter("pipeline_traces_dropped"),
pipelineTracesRetained: factory.NewCounter("pipeline_traces_retained"),
@@ -575,6 +588,7 @@ func (qs *queueSupplier) newMetrics(p common.Position) (storage.Metrics, observa
totalFilePartBytes: factory.NewGauge("total_file_part_bytes", common.ShardLabelNames()...),
totalFilePartUncompressedBytes: factory.NewGauge("total_file_part_uncompressed_bytes", common.ShardLabelNames()...),
pendingDataCount: factory.NewGauge("pending_data_count", common.ShardLabelNames()...),
+ mergeQuarantinedParts: factory.NewGauge("merge_quarantined_parts", common.ShardLabelNames()...),
},
indexMetrics: inverted.NewMetrics(factory, common.SegLabelNames()...),
}, factory
@@ -619,6 +633,16 @@ func (tst *tsTable) Collect(m storage.Metrics) {
metrics.totalFileBlocks.Set(float64(totalFileBlocks), tst.p.ShardLabelValues()...)
metrics.totalFilePartBytes.Set(float64(totalFilePartBytes), tst.p.ShardLabelValues()...)
metrics.totalFilePartUncompressedBytes.Set(float64(totalFilePartUncompressedBytes), tst.p.ShardLabelValues()...)
+
+ tst.quarantineMu.Lock()
+ var quarantinedParts int
+ for _, fails := range tst.quarantineFails {
+ if fails >= quarantineThreshold {
+ quarantinedParts++
+ }
+ }
+ tst.quarantineMu.Unlock()
+ metrics.mergeQuarantinedParts.Set(float64(quarantinedParts), tst.p.ShardLabelValues()...)
}
func (tst *tsTable) deleteMetrics() {
@@ -636,6 +660,7 @@ func (tst *tsTable) deleteMetrics() {
tst.metrics.tbMetrics.totalFilePartBytes.Delete(tst.p.ShardLabelValues()...)
tst.metrics.tbMetrics.totalFilePartUncompressedBytes.Delete(tst.p.ShardLabelValues()...)
tst.metrics.tbMetrics.pendingDataCount.Delete(tst.p.ShardLabelValues()...)
+ tst.metrics.tbMetrics.mergeQuarantinedParts.Delete(tst.p.ShardLabelValues()...)
tst.metrics.indexMetrics.DeleteAll(tst.p.SegLabelValues()...)
}
@@ -653,4 +678,6 @@ type tbMetrics struct {
totalFilePartUncompressedBytes meter.Gauge
pendingDataCount meter.Gauge
+
+ mergeQuarantinedParts meter.Gauge
}
diff --git a/banyand/trace/part_iter.go b/banyand/trace/part_iter.go
index 23b582d8e..2807d9d74 100644
--- a/banyand/trace/part_iter.go
+++ b/banyand/trace/part_iter.go
@@ -260,7 +260,9 @@ type partMergeIter struct {
primaryBlockMetadata []primaryBlockMetadata
compressedPrimaryBuf []byte
primaryBuf []byte
+ partPath string
block blockPointer
+ partID uint64
primaryMetadataIdx int
}
@@ -273,6 +275,8 @@ func (pmi *partMergeIter) reset() {
pmi.primaryBuf = pmi.primaryBuf[:0]
pmi.compressedPrimaryBuf = pmi.compressedPrimaryBuf[:0]
pmi.block.reset()
+ pmi.partID = 0
+ pmi.partPath = ""
}
func (pmi *partMergeIter) mustInitFromPart(p *part) {
@@ -280,6 +284,8 @@ func (pmi *partMergeIter) mustInitFromPart(p *part) {
pmi.seqReaders.init(p)
pmi.tagType = p.tagType
pmi.primaryBlockMetadata = p.primaryBlockMetadata
+ pmi.partID = p.partMetadata.ID
+ pmi.partPath = p.path
}
func (pmi *partMergeIter) error() error {
@@ -296,17 +302,27 @@ func (pmi *partMergeIter) nextBlockMetadata() bool {
pmi.block.reset()
if len(pmi.primaryBuf) == 0 {
if err := pmi.loadPrimaryBuf(); err != nil {
- pmi.err = err
+ pmi.err = pmi.attributeReadErr(err)
return false
}
}
if err := pmi.loadBlockMetadata(); err != nil {
- pmi.err = err
+ pmi.err = pmi.attributeReadErr(err)
return false
}
return true
}
+// attributeReadErr wraps a read-side failure in *unreadablePartError so the merge loop can
+// recover the failing part's identity. io.EOF marks a clean end of iteration, not a failure,
+// so it is left unwrapped: error() filters it out.
+func (pmi *partMergeIter) attributeReadErr(err error) error {
+ if errors.Is(err, io.EOF) {
+ return err
+ }
+ return &unreadablePartError{err: err, partID: pmi.partID, partPath: pmi.partPath}
+}
+
func (pmi *partMergeIter) loadPrimaryBuf() error {
if pmi.primaryMetadataIdx >= len(pmi.primaryBlockMetadata) {
return io.EOF
diff --git a/banyand/trace/quarantine.go b/banyand/trace/quarantine.go
new file mode 100644
index 000000000..c76189642
--- /dev/null
+++ b/banyand/trace/quarantine.go
@@ -0,0 +1,85 @@
+// Licensed to Apache Software Foundation (ASF) under one or more contributor
+// license agreements. See the NOTICE file distributed with
+// this work for additional information regarding copyright
+// ownership. Apache Software Foundation (ASF) licenses this file to you under
+// the Apache License, Version 2.0 (the "License"); you may
+// not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+package trace
+
+import "errors"
+
+// quarantineThreshold is the number of consecutive attributable read failures a part
+// must accumulate before merge selection excludes it. It tolerates transient IO errors
+// while a deterministic decode failure crosses it within three retry cycles.
+const quarantineThreshold = 3
+
+// recordUnreadablePart inspects mergeErr for an attributable *unreadablePartError and, if
+// found, increments that part's consecutive-failure count. It returns true when mergeErr
+// was attributed to a part, regardless of whether the count crossed the quarantine
+// threshold on this call. Crossing the threshold logs one Error line and increments the
+// quarantine metric.
+func (tst *tsTable) recordUnreadablePart(mergeErr error) bool {
+ var unreadableErr *unreadablePartError
+ if !errors.As(mergeErr, &unreadableErr) {
+ return false
+ }
+
+ tst.quarantineMu.Lock()
+ if tst.quarantineFails == nil {
+ tst.quarantineFails = make(map[uint64]int)
+ }
+ tst.quarantineFails[unreadableErr.partID]++
+ fails := tst.quarantineFails[unreadableErr.partID]
+ tst.quarantineMu.Unlock()
+
+ if fails == quarantineThreshold {
+ tst.l.Error().
+ Str("group", tst.group).
+ Uint32("shard", uint32(tst.shardID)).
+ Uint64("partID", unreadableErr.partID).
+ Str("partPath", unreadableErr.partPath).
+ Err(unreadableErr.err).
+ Msg("part crossed the quarantine threshold; excluding it from merge selection")
+ tst.incTotalMergePartQuarantined(1)
+ }
+ return true
+}
+
+// isPartQuarantined reports whether partID has crossed the quarantine threshold and
+// should be excluded from merge selection.
+func (tst *tsTable) isPartQuarantined(partID uint64) bool {
+ tst.quarantineMu.Lock()
+ defer tst.quarantineMu.Unlock()
+ return tst.quarantineFails[partID] >= quarantineThreshold
+}
+
+// hasQuarantinedParts reports whether the registry currently holds any entries, so
+// callers on the hot selection path can skip building a liveIDs set when it is empty.
+func (tst *tsTable) hasQuarantinedParts() bool {
+ tst.quarantineMu.Lock()
+ defer tst.quarantineMu.Unlock()
+ return len(tst.quarantineFails) > 0
+}
+
+// sweepQuarantine removes registry entries for parts no longer present in liveIDs
+// (merged away, TTL'd, or deleted). The registry is in-memory only.
+func (tst *tsTable) sweepQuarantine(liveIDs map[uint64]struct{}) {
+ tst.quarantineMu.Lock()
+ defer tst.quarantineMu.Unlock()
+ for partID := range tst.quarantineFails {
+ if _, live := liveIDs[partID]; !live {
+ delete(tst.quarantineFails, partID)
+ }
+ }
+}
diff --git a/banyand/trace/quarantine_test.go b/banyand/trace/quarantine_test.go
new file mode 100644
index 000000000..cf428f09f
--- /dev/null
+++ b/banyand/trace/quarantine_test.go
@@ -0,0 +1,184 @@
+// Licensed to Apache Software Foundation (ASF) under one or more contributor
+// license agreements. See the NOTICE file distributed with
+// this work for additional information regarding copyright
+// ownership. Apache Software Foundation (ASF) licenses this file to you under
+// the Apache License, Version 2.0 (the "License"); you may
+// not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+package trace
+
+import (
+ "errors"
+ "os"
+ "path/filepath"
+ "testing"
+
+ "github.com/stretchr/testify/require"
+
+ "github.com/apache/skywalking-banyandb/banyand/protector"
+ "github.com/apache/skywalking-banyandb/pkg/fs"
+ "github.com/apache/skywalking-banyandb/pkg/logger"
+ "github.com/apache/skywalking-banyandb/pkg/test"
+)
+
+// corruptPrimaryBin overwrites partID's primary.bin with same-length garbage so the
+// reader's mustReadFull still succeeds and only zstd decompression fails, producing a
+// plain read-side error rather than a panic (mirrors merger_failure_cleanup_test.go).
+func corruptPrimaryBin(t *testing.T, root string, partID uint64) {
+ t.Helper()
+ primaryPath := filepath.Join(partPath(root, partID), primaryFilename)
+ data, readErr := os.ReadFile(primaryPath)
+ require.NoError(t, readErr)
+ require.NotEmpty(t, data)
+ garbage := make([]byte, len(data))
+ for i := range garbage {
+ garbage[i] = byte(0xAA ^ i)
+ }
+ require.NoError(t, os.WriteFile(primaryPath, garbage, 0o600))
+}
+
+// flushFilePart flushes tsData into a fresh, unopened file part at partID so a test can
+// corrupt its files on disk before the part is ever opened.
+func flushFilePart(t *testing.T, fileSystem fs.FileSystem, root string, partID uint64, tsData *traces) {
+ t.Helper()
+ mp := generateMemPart()
+ mp.mustInitFromTraces(tsData)
+ mp.mustFlush(fileSystem, partPath(root, partID))
+ releaseMemPart(mp)
+}
+
+// mustCreateFilePart flushes tsData into a fresh file part at partID and opens it.
+func mustCreateFilePart(t *testing.T, fileSystem fs.FileSystem, root string, partID uint64, tsData *traces) *part {
+ t.Helper()
+ flushFilePart(t, fileSystem, root, partID, tsData)
+ p := mustOpenFilePart(partID, root, fileSystem)
+ p.partMetadata.ID = partID
+ return p
+}
+
+// TestUnreadablePartErrorAttribution reproduces a merge over one corrupt and one healthy
+// file part and asserts the resulting error identifies the corrupt part via
+// *unreadablePartError, recovered through errors.As across the block_reader/mergeBlocks
+// wrapping chain.
+func TestUnreadablePartErrorAttribution(t *testing.T) {
+ tmpPath, defFn := test.Space(require.New(t))
+ defer defFn()
+ fileSystem := fs.NewLocalFileSystem()
+
+ const badID, goodID = 1, 2
+ flushFilePart(t, fileSystem, tmpPath, badID, ts)
+ corruptPrimaryBin(t, tmpPath, badID)
+ // Open after corruption: metadata/tag files are untouched, only primary.bin is
+ // garbage, so mustOpenFilePart still succeeds and the failure surfaces on read.
+ pBad := mustOpenFilePart(badID, tmpPath, fileSystem)
+ pBad.partMetadata.ID = badID
+ defer pBad.close()
+
+ pGood := mustCreateFilePart(t, fileSystem, tmpPath, goodID, ts)
+ defer pGood.close()
+
+ tst := &tsTable{pm: protector.Nop{}, fileSystem: fileSystem, root: tmpPath, l: logger.GetLogger("test")}
+ closeCh := make(chan struct{})
+ defer close(closeCh)
+
+ parts := []*partWrapper{newPartWrapper(nil, pBad), newPartWrapper(nil, pGood)}
+ _, _, mergeErr := tst.mergeParts(fileSystem, closeCh, parts, 100, tmpPath, nil, nil)
+ require.Error(t, mergeErr)
+
+ var unreadableErr *unreadablePartError
+ require.True(t, errors.As(mergeErr, &unreadableErr), "expected *unreadablePartError in the error chain, got: %v", mergeErr)
+ require.Equal(t, uint64(badID), unreadableErr.partID)
+ require.Equal(t, pBad.path, unreadableErr.partPath)
+}
+
+// TestMergeQuarantinesUnreadablePart simulates the retry loop hitting the same poison
+// part three times, then asserts selection excludes it while still surfacing the
+// healthy backlog for merging.
+func TestMergeQuarantinesUnreadablePart(t *testing.T) {
+ tmpPath, defFn := test.Space(require.New(t))
+ defer defFn()
+ fileSystem := fs.NewLocalFileSystem()
+
+ const badID = uint64(500)
+ flushFilePart(t, fileSystem, tmpPath, badID, tsTS1)
+ corruptPrimaryBin(t, tmpPath, badID)
+ pBad := mustOpenFilePart(badID, tmpPath, fileSystem)
+ pBad.partMetadata.ID = badID
+ // badPW's initial ref is later transferred to the snapshot built below; decRef is
+ // not called here to avoid a double-release once the snapshot drops its ref.
+ badPW := newPartWrapper(nil, pBad)
+
+ var healthyPWs []*partWrapper
+ for _, partID := range []uint64{501, 502, 503} {
+ p := mustCreateFilePart(t, fileSystem, tmpPath, partID, tsTS1)
+ healthyPWs = append(healthyPWs, newPartWrapper(nil, p))
+ }
+
+ tst := &tsTable{
+ pm: protector.Nop{},
+ fileSystem: fileSystem,
+ root: tmpPath,
+ l: logger.GetLogger("test"),
+ curPartID: 1000,
+ option: option{
+ mergePolicy: newDefaultMergePolicyForTesting(),
+ },
+ }
+ closeCh := make(chan struct{})
+ defer close(closeCh)
+
+ // Produce the real attribution error once, then simulate three independent retry
+ // cycles hitting the same poison part.
+ _, _, mergeErr := tst.mergeParts(fileSystem, closeCh, []*partWrapper{badPW, healthyPWs[0]}, 9000, tmpPath, nil, nil)
+ require.Error(t, mergeErr)
+ var unreadableErr *unreadablePartError
+ require.True(t, errors.As(mergeErr, &unreadableErr))
+ require.Equal(t, badID, unreadableErr.partID)
+
+ require.False(t, tst.isPartQuarantined(badID), "should not be quarantined before crossing the threshold")
+ for attempt := 1; attempt <= quarantineThreshold; attempt++ {
+ attributed := tst.recordUnreadablePart(mergeErr)
+ require.True(t, attributed, "attempt %d: expected mergeErr to attribute to a part", attempt)
+ }
+ require.True(t, tst.isPartQuarantined(badID), "part should be quarantined after %d consecutive failures", quarantineThreshold)
+
+ allParts := append([]*partWrapper{badPW}, healthyPWs...)
+ snp := &snapshot{parts: allParts, epoch: 1}
+ snp.incRef()
+ defer snp.decRef()
+
+ dst, toBeMerged := tst.getPartsToMergeUpTo(snp, uint64(1<<30), nil, 0)
+ require.GreaterOrEqual(t, len(dst), 2, "healthy backlog should still be selectable")
+ for _, pw := range dst {
+ require.NotEqual(t, badID, pw.ID(), "quarantined part must be excluded from selection")
+ }
+ _, badSelected := toBeMerged[badID]
+ require.False(t, badSelected, "quarantined part must not appear in toBeMerged")
+}
+
+// TestQuarantineSweep asserts sweepQuarantine drops registry entries for parts no
+// longer present in the live set (merged away, TTL'd, or deleted).
+func TestQuarantineSweep(t *testing.T) {
+ tst := &tsTable{l: logger.GetLogger("test")}
+ tst.quarantineFails = map[uint64]int{
+ 42: quarantineThreshold,
+ 43: quarantineThreshold,
+ }
+ require.True(t, tst.isPartQuarantined(42))
+ require.True(t, tst.isPartQuarantined(43))
+
+ tst.sweepQuarantine(map[uint64]struct{}{43: {}})
+
+ require.False(t, tst.isPartQuarantined(42), "42 should be swept: absent from liveIDs")
+ require.True(t, tst.isPartQuarantined(43), "43 should survive the sweep: present in liveIDs")
+}
diff --git a/banyand/trace/tstable.go b/banyand/trace/tstable.go
index b00621c5f..ba86ab80e 100644
--- a/banyand/trace/tstable.go
+++ b/banyand/trace/tstable.go
@@ -88,6 +88,10 @@ type tsTable struct {
// introduction path (atomic add only), reset by a successful finalize round.
unsampledBytes atomic.Int64
inFlightMu sync.RWMutex
+ // quarantineFails counts consecutive attributable read failures per partID (Fix B).
+ // Lazily initialized; guarded by quarantineMu.
+ quarantineFails map[uint64]int
+ quarantineMu sync.Mutex
sync.RWMutex
shardID common.ShardID
isHot bool
diff --git a/banyand/trace/unreadable_part.go b/banyand/trace/unreadable_part.go
new file mode 100644
index 000000000..a93016ccf
--- /dev/null
+++ b/banyand/trace/unreadable_part.go
@@ -0,0 +1,35 @@
+// Licensed to Apache Software Foundation (ASF) under one or more contributor
+// license agreements. See the NOTICE file distributed with
+// this work for additional information regarding copyright
+// ownership. Apache Software Foundation (ASF) licenses this file to you under
+// the Apache License, Version 2.0 (the "License"); you may
+// not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+package trace
+
+import "fmt"
+
+// unreadablePartError marks a merge-read failure attributable to one input part.
+type unreadablePartError struct {
+ err error
+ partPath string
+ partID uint64
+}
+
+func (e *unreadablePartError) Error() string {
+ return fmt.Sprintf("part %016x (%s) is unreadable: %s", e.partID, e.partPath, e.err)
+}
+
+func (e *unreadablePartError) Unwrap() error {
+ return e.err
+}
(skywalking-banyandb) 01/06: fix(storage): remove partially-written merge output on failure
Posted by ha...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
hanahmily pushed a commit to branch trace-merge-resilience
in repository https://gitbox.apache.org/repos/asf/skywalking-banyandb.git
commit 241240593a6db9d3f56e4e00971a1c978fdec1a4
Author: owenwillison-vajra <30...@qq.com>
AuthorDate: Sat Aug 8 13:15:53 2026 +0000
fix(storage): remove partially-written merge output on failure
A merge that fails after the block writer created the output part
directory previously leaked that directory on disk. Under a persistent
merge failure this leaks one orphan directory per attempt and exhausts
the volume's inodes. Remove the output directory on any error return or
panic in trace, stream, measure, and sidx mergeParts.
---
banyand/internal/sidx/merge.go | 13 ++++
banyand/measure/merger.go | 13 ++++
banyand/stream/merger.go | 13 ++++
banyand/trace/merger.go | 13 ++++
banyand/trace/merger_failure_cleanup_test.go | 96 ++++++++++++++++++++++++++++
5 files changed, 148 insertions(+)
diff --git a/banyand/internal/sidx/merge.go b/banyand/internal/sidx/merge.go
index c8452511b..267a5b719 100644
--- a/banyand/internal/sidx/merge.go
+++ b/banyand/internal/sidx/merge.go
@@ -97,7 +97,19 @@ func (s *sidx) mergeParts(fileSystem fs.FileSystem, closeCh <-chan struct{}, par
br := generateBlockReader()
br.init(pii)
bw := generateBlockWriter()
+ outputPublished := false
bw.mustInitForFilePart(fileSystem, dstPath, shouldCache)
+ // Remove the partially-written output on any failure or panic so failed merges never leak part directories.
+ defer func() {
+ if outputPublished {
+ return
+ }
+ if panicVal := recover(); panicVal != nil {
+ fileSystem.MustRMAll(dstPath)
+ panic(panicVal)
+ }
+ fileSystem.MustRMAll(dstPath)
+ }()
pm, err := mergeBlocks(closeCh, bw, br, conflictTags, keep)
releaseBlockWriter(bw)
@@ -134,6 +146,7 @@ func (s *sidx) mergeParts(fileSystem fs.FileSystem, closeCh <-chan struct{}, par
// No SyncPath: mustWriteMetadata goes through MustFlushAtomic which
// already fsyncs the parent directory after rename.
p := mustOpenPart(partID, dstPath, fileSystem)
+ outputPublished = true
return newPartWrapper(nil, p), nil
}
diff --git a/banyand/measure/merger.go b/banyand/measure/merger.go
index de3693289..a335bd0c6 100644
--- a/banyand/measure/merger.go
+++ b/banyand/measure/merger.go
@@ -251,7 +251,19 @@ func (tst *tsTable) mergeParts(fileSystem fs.FileSystem, closeCh <-chan struct{}
br := generateBlockReader()
br.init(pii)
bw := generateBlockWriter()
+ outputPublished := false
bw.mustInitForFilePart(fileSystem, dstPath, shouldCache)
+ // Remove the partially-written output on any failure or panic so failed merges never leak part directories.
+ defer func() {
+ if outputPublished {
+ return
+ }
+ if panicVal := recover(); panicVal != nil {
+ fileSystem.MustRMAll(dstPath)
+ panic(panicVal)
+ }
+ fileSystem.MustRMAll(dstPath)
+ }()
pm, mergedTagType, err := mergeBlocks(closeCh, bw, br, conflictColumns)
releaseBlockWriter(bw)
@@ -267,6 +279,7 @@ func (tst *tsTable) mergeParts(fileSystem fs.FileSystem, closeCh <-chan struct{}
// No SyncPath: mustWriteMetadata goes through WriteAtomic which already
// fsyncs the parent directory after rename.
p := mustOpenFilePart(partID, root, fileSystem)
+ outputPublished = true
return newPartWrapper(nil, p), nil
}
diff --git a/banyand/stream/merger.go b/banyand/stream/merger.go
index 320308a90..a55989cbb 100644
--- a/banyand/stream/merger.go
+++ b/banyand/stream/merger.go
@@ -249,7 +249,19 @@ func (tst *tsTable) mergeParts(fileSystem fs.FileSystem, closeCh <-chan struct{}
br := generateBlockReader()
br.init(pii)
bw := generateBlockWriter()
+ outputPublished := false
bw.mustInitForFilePart(fileSystem, dstPath, shouldCache)
+ // Remove the partially-written output on any failure or panic so failed merges never leak part directories.
+ defer func() {
+ if outputPublished {
+ return
+ }
+ if panicVal := recover(); panicVal != nil {
+ fileSystem.MustRMAll(dstPath)
+ panic(panicVal)
+ }
+ fileSystem.MustRMAll(dstPath)
+ }()
pm, mergedTagType, err := mergeBlocks(closeCh, bw, br, conflictTags)
releaseBlockWriter(bw)
@@ -265,6 +277,7 @@ func (tst *tsTable) mergeParts(fileSystem fs.FileSystem, closeCh <-chan struct{}
// No SyncPath: mustWriteMetadata goes through WriteAtomic which already
// fsyncs the parent directory after rename.
p := mustOpenFilePart(partID, root, fileSystem)
+ outputPublished = true
return newPartWrapper(nil, p), nil
}
diff --git a/banyand/trace/merger.go b/banyand/trace/merger.go
index f4ff236c1..ae71470c8 100644
--- a/banyand/trace/merger.go
+++ b/banyand/trace/merger.go
@@ -1042,7 +1042,19 @@ func (tst *tsTable) mergeParts(fileSystem fs.FileSystem, closeCh <-chan struct{}
br := generateBlockReader()
br.init(pii)
bw := generateBlockWriter()
+ outputPublished := false
bw.mustInitForFilePart(fileSystem, dstPath, shouldCache, int(traceSize))
+ // Remove the partially-written output on any failure or panic so failed merges never leak part directories.
+ defer func() {
+ if outputPublished {
+ return
+ }
+ if panicVal := recover(); panicVal != nil {
+ fileSystem.MustRMAll(dstPath)
+ panic(panicVal)
+ }
+ fileSystem.MustRMAll(dstPath)
+ }()
conflictTags := collectConflictTags(parts)
var minTimestamp, maxTimestamp int64
@@ -1097,6 +1109,7 @@ func (tst *tsTable) mergeParts(fileSystem fs.FileSystem, closeCh <-chan struct{}
// which already fsyncs the parent directory after rename. The last atomic
// metadata write covers all prior dirent changes (data file creations).
p := mustOpenFilePart(partID, root, fileSystem)
+ outputPublished = true
return newPartWrapper(nil, p), dropped, nil
}
diff --git a/banyand/trace/merger_failure_cleanup_test.go b/banyand/trace/merger_failure_cleanup_test.go
new file mode 100644
index 000000000..98879fa28
--- /dev/null
+++ b/banyand/trace/merger_failure_cleanup_test.go
@@ -0,0 +1,96 @@
+// Licensed to Apache Software Foundation (ASF) under one or more contributor
+// license agreements. See the NOTICE file distributed with
+// this work for additional information regarding copyright
+// ownership. Apache Software Foundation (ASF) licenses this file to you under
+// the Apache License, Version 2.0 (the "License"); you may
+// not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+package trace
+
+import (
+ "os"
+ "path/filepath"
+ "testing"
+
+ "github.com/stretchr/testify/require"
+
+ "github.com/apache/skywalking-banyandb/banyand/protector"
+ "github.com/apache/skywalking-banyandb/pkg/fs"
+ "github.com/apache/skywalking-banyandb/pkg/test"
+)
+
+// countTopLevelEntries returns the number of top-level directory entries under
+// root, so a test can assert that a failed merge left no new part directory.
+func countTopLevelEntries(t *testing.T, root string) int {
+ t.Helper()
+ entries, err := os.ReadDir(root)
+ require.NoError(t, err)
+ return len(entries)
+}
+
+// TestMergePartsCleansOutputOnFailure reproduces the 2026-08-07 showcase outage
+// mechanism: mergeParts creates the output part directory before mergeBlocks
+// runs. Before Fix A, a corrupt input part left an orphan output directory on
+// every failed merge attempt (~430k dirs/shard in production). This asserts
+// the merge fails but leaves the shard root exactly as it was.
+func TestMergePartsCleansOutputOnFailure(t *testing.T) {
+ tmpPath, defFn := test.Space(require.New(t))
+ defer defFn()
+ fileSystem := fs.NewLocalFileSystem()
+
+ mp1 := generateMemPart()
+ mp1.mustInitFromTraces(ts)
+ mp1.mustFlush(fileSystem, partPath(tmpPath, 1))
+ releaseMemPart(mp1)
+
+ mp2 := generateMemPart()
+ mp2.mustInitFromTraces(ts)
+ mp2.mustFlush(fileSystem, partPath(tmpPath, 2))
+ releaseMemPart(mp2)
+
+ // Corrupt part 1's primary.bin in place (same length, so the reader's
+ // mustReadFull still succeeds and only zstd decompression fails), which
+ // makes mergeBlocks return a plain error rather than panic.
+ primaryPath := filepath.Join(partPath(tmpPath, 1), primaryFilename)
+ data, readErr := os.ReadFile(primaryPath)
+ require.NoError(t, readErr)
+ require.NotEmpty(t, data)
+ garbage := make([]byte, len(data))
+ for i := range garbage {
+ garbage[i] = byte(0xAA ^ i)
+ }
+ require.NoError(t, os.WriteFile(primaryPath, garbage, 0o600))
+
+ p1 := mustOpenFilePart(1, tmpPath, fileSystem)
+ p1.partMetadata.ID = 1
+ p2 := mustOpenFilePart(2, tmpPath, fileSystem)
+ p2.partMetadata.ID = 2
+ defer p1.close()
+ defer p2.close()
+
+ tst := &tsTable{pm: protector.Nop{}, fileSystem: fileSystem, root: tmpPath}
+ closeCh := make(chan struct{})
+ defer close(closeCh)
+
+ beforeCount := countTopLevelEntries(t, tmpPath)
+
+ for attempt, partID := range []uint64{99, 100, 101} {
+ parts := []*partWrapper{newPartWrapper(nil, p1), newPartWrapper(nil, p2)}
+ _, _, mergeErr := tst.mergeParts(fileSystem, closeCh, parts, partID, tmpPath, nil, nil)
+ require.Errorf(t, mergeErr, "attempt %d: expected merge to fail on corrupt primary.bin", attempt)
+
+ afterCount := countTopLevelEntries(t, tmpPath)
+ require.Equalf(t, beforeCount, afterCount,
+ "attempt %d: shard root entry count changed (leaked output dir for partID %d)", attempt, partID)
+ }
+}
(skywalking-banyandb) 04/06: fix(trace): survive panics in the merge workers and dispatcher
Posted by ha...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
hanahmily pushed a commit to branch trace-merge-resilience
in repository https://gitbox.apache.org/repos/asf/skywalking-banyandb.git
commit c8f4f4c4e248f9f22e34c7d34ce064c7d5aa0efe
Author: owenwillison-vajra <30...@qq.com>
AuthorDate: Sat Aug 8 13:55:37 2026 +0000
fix(trace): survive panics in the merge workers and dispatcher
A panic inside a merge (e.g. a disk-full logger.Panicf in the fs layer)
previously killed the lane worker goroutine: run.Go recovers and counts
the panic but never respawns the worker, so once all fast-lane workers
died the dispatcher blocked forever on the lane channel and the
in-flight parts stayed pinned. Convert merge-execution panics into
ordinary merge errors at the worker so the epilogue (semaphore release,
in-flight unpin, failure accounting, backoff) always runs, and protect
the dispatcher's selection cycle the same way. New metric:
total_merge_panic_recovered.
---
banyand/trace/merge_panic_test.go | 196 ++++++++++++++++++++++++++++++++++++++
banyand/trace/merger.go | 43 ++++++---
banyand/trace/metrics.go | 13 +++
3 files changed, 241 insertions(+), 11 deletions(-)
diff --git a/banyand/trace/merge_panic_test.go b/banyand/trace/merge_panic_test.go
new file mode 100644
index 000000000..43cfd486b
--- /dev/null
+++ b/banyand/trace/merge_panic_test.go
@@ -0,0 +1,196 @@
+// Licensed to the Apache Software Foundation (ASF) under one or more contributor
+// license agreements. See the NOTICE file distributed with this work for
+// additional information regarding copyright ownership. The ASF licenses this
+// file to you under the Apache License, Version 2.0 (the "License"); you may not
+// use this file except in compliance with the License. You may obtain a copy of
+// the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
+// WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
+// License for the specific language governing permissions and limitations
+// under the License.
+
+package trace
+
+import (
+ "context"
+ "os"
+ "path/filepath"
+ "testing"
+ "time"
+
+ "github.com/stretchr/testify/require"
+
+ "github.com/apache/skywalking-banyandb/api/common"
+ "github.com/apache/skywalking-banyandb/banyand/protector"
+ "github.com/apache/skywalking-banyandb/pkg/fs"
+ "github.com/apache/skywalking-banyandb/pkg/logger"
+ "github.com/apache/skywalking-banyandb/pkg/test"
+ "github.com/apache/skywalking-banyandb/pkg/timestamp"
+)
+
+// corruptSpansBin truncates partID's spans.bin by one byte so the sole block's declared
+// size no longer matches the file: seqReader.mustReadFull's io.ReadFull hits a short read
+// (io.ErrUnexpectedEOF, not io.EOF) and panics via logger.Panicf("cannot read data: ...")
+// instead of surfacing a plain error. meta.bin, primary.bin and metadata.json are
+// untouched, so mustOpenFilePart still succeeds; the panic fires only once the merge
+// actually reads spans.bin, reproducing the outage's "panic inside the merge path" shape
+// without a production test hook.
+func corruptSpansBin(t *testing.T, root string, partID uint64) {
+ t.Helper()
+ spansPath := filepath.Join(partPath(root, partID), spansFilename)
+ data, readErr := os.ReadFile(spansPath)
+ require.NoError(t, readErr)
+ require.Greater(t, len(data), 1, "spans.bin must have data to truncate meaningfully")
+ require.NoError(t, os.WriteFile(spansPath, data[:len(data)-1], 0o600))
+}
+
+// singleTraceSet builds a minimal traces batch with one trace, so each test part has an
+// identity distinct from every other part (avoids incidental traceID overlap across the
+// parts merged in this test).
+func singleTraceSet(traceID string, ts int64) *traces {
+ return &traces{
+ traceIDs: []string{traceID},
+ timestamps: []int64{ts},
+ spanIDs: []string{"span-" + traceID},
+ spans: [][]byte{[]byte("payload-" + traceID)},
+ tags: [][]*tagValue{{}},
+ }
+}
+
+// waitForMergeControlChange blocks until condition() is true, waking only on
+// tst.mergeControl's change notifications (never a bare sleep), and fails the test if
+// deadline elapses first.
+func waitForMergeControlChange(t *testing.T, tst *tsTable, deadline time.Duration, condition func() bool) {
+ t.Helper()
+ timeout := time.After(deadline)
+ for {
+ if condition() {
+ return
+ }
+ state := tst.mergeControl.state()
+ select {
+ case <-state.changed:
+ case <-timeout:
+ t.Fatalf("timed out after %s waiting for merge control condition", deadline)
+ }
+ }
+}
+
+func consecutiveMergeFailures(tst *tsTable) int {
+ tst.mergeControl.mu.Lock()
+ defer tst.mergeControl.mu.Unlock()
+ return tst.mergeControl.consecutiveFailures
+}
+
+// TestMergeWorkerSurvivesPanic reproduces the third defect behind the 2026-08-07 showcase
+// outage: before Fix D, a panic inside the merge invocation (there: mkdir under inode
+// exhaustion) killed the lane worker goroutine outright -- run.Go recovers the panic at
+// the outermost layer and counts it, but the goroutine exits and is never respawned. With
+// every fast-lane worker dead, the dispatcher blocks forever writing to the lane channel
+// and the in-flight request's parts stay pinned in tst.inFlight.
+//
+// This drives the real mergeLoop/dispatcherLoop/mergeLaneWorker machinery started by
+// newTSTable (via triggerMerge), not a direct mergeParts call, so the assertions exercise
+// the worker's recover-to-error wrapping itself rather than just mergeParts' own panic-safe
+// cleanup (already covered by TestMergePartsCleansOutputOnFailure).
+func TestMergeWorkerSurvivesPanic(t *testing.T) {
+ fileSystem := fs.NewLocalFileSystem()
+ tmpPath, cleanup := test.Space(require.New(t))
+ t.Cleanup(cleanup)
+
+ tableRoot := filepath.Join(tmpPath, "table")
+ fileSystem.MkdirPanicIfExist(tableRoot, 0o755)
+ tst, tableErr := newTSTable(
+ fileSystem,
+ tableRoot,
+ common.Position{Database: "merge-panic-survival"},
+ logger.GetLogger("merge-panic-survival"),
+ timestamp.TimeRange{},
+ option{
+ flushTimeout: 0,
+ mergePolicy: newDefaultMergePolicyForTesting(),
+ protector: protector.Nop{},
+ },
+ nil,
+ )
+ require.NoError(t, tableErr)
+ t.Cleanup(func() { require.NoError(t, tst.Close()) })
+
+ const badID, goodID = uint64(1), uint64(2)
+ tst.observePartID(goodID)
+ flushFilePart(t, fileSystem, tableRoot, badID, singleTraceSet("trace-bad", 1))
+ corruptSpansBin(t, tableRoot, badID)
+ flushFilePart(t, fileSystem, tableRoot, goodID, singleTraceSet("trace-good", 2))
+
+ tst.mustAddFilePart(badID, nil)
+ tst.mustAddFilePart(goodID, nil)
+
+ require.NoError(t, tst.triggerMerge())
+
+ // The panicking merge must be recovered into an ordinary failure: recordOutcome(false)
+ // increments consecutiveFailures. A regression that lets the panic kill the worker
+ // goroutine would never notify mergeControl again, so this wait times out instead of
+ // hanging the whole suite.
+ waitForMergeControlChange(t, tst, 10*time.Second, func() bool {
+ return consecutiveMergeFailures(tst) >= 1
+ })
+
+ // The worker epilogue (releaseDispatchRequest) must still run unconditionally after a
+ // recovered panic, unpinning both parts from tst.inFlight -- the exact pin leak the
+ // outage hit when panics killed the goroutine before the epilogue could execute.
+ waitForMergeControlChange(t, tst, 10*time.Second, tst.mergeInFlightEmpty)
+
+ // Quarantine the poison part directly (whitebox, same package) instead of relying on
+ // Fix B's attribution: a recovered panic is an *unattributed* failure by design (it
+ // never flows through *unreadablePartError), so backoff -- not quarantine -- is its
+ // normal safety net. Excluding badID here isolates the second trigger below to the
+ // healthy parts, so a successful merge is unambiguous proof the lane worker survived.
+ tst.quarantineMu.Lock()
+ if tst.quarantineFails == nil {
+ tst.quarantineFails = make(map[uint64]int)
+ }
+ tst.quarantineFails[badID] = quarantineThreshold
+ tst.quarantineMu.Unlock()
+
+ const extraID1, extraID2 = uint64(3), uint64(4)
+ tst.observePartID(extraID2)
+ flushFilePart(t, fileSystem, tableRoot, extraID1, singleTraceSet("trace-extra-1", 3))
+ flushFilePart(t, fileSystem, tableRoot, extraID2, singleTraceSet("trace-extra-2", 4))
+ tst.mustAddFilePart(extraID1, nil)
+ tst.mustAddFilePart(extraID2, nil)
+
+ require.NoError(t, tst.triggerMerge())
+ waitCtx, cancel := context.WithTimeout(context.Background(), 15*time.Second)
+ defer cancel()
+ require.NoError(t, tst.waitForMergeIdle(waitCtx), "the lane worker must still be alive to serve this second merge")
+
+ snap := tst.currentSnapshot()
+ require.NotNil(t, snap)
+ defer snap.decRef()
+
+ var sawBadID bool
+ mergedCount := 0
+ for _, pw := range snap.parts {
+ if pw.ID() == badID {
+ sawBadID = true
+ continue
+ }
+ mergedCount++
+ }
+ require.True(t, sawBadID, "the quarantined poison part must remain in the snapshot, unmerged")
+ require.Equal(t, 1, mergedCount, "the three healthy parts (goodID + two extras) must have merged into one output part")
+
+ tst.inFlightMu.RLock()
+ require.Empty(t, tst.inFlight)
+ tst.inFlightMu.RUnlock()
+
+ // tst.metrics is nil for this lightweight table construction (no metrics supplier
+ // wired), matching every other whitebox test in this file (e.g.
+ // TestMergeQuarantinesUnreadablePart); incTotalMergePanicRecovered's nil guard makes
+ // that safe, so the counter is not asserted numerically here. Its wiring (struct
+ // field, DeleteAll, both newMetrics constructors) is exercised by the package build.
+}
diff --git a/banyand/trace/merger.go b/banyand/trace/merger.go
index c97a2a858..9f79a9e06 100644
--- a/banyand/trace/merger.go
+++ b/banyand/trace/merger.go
@@ -22,6 +22,7 @@ import (
"errors"
"fmt"
"path/filepath"
+ "runtime/debug"
"sort"
"sync"
"sync/atomic"
@@ -341,19 +342,30 @@ func (tst *tsTable) dispatcherLoop(threshold uint64, fastCh, slowCh chan *mergeD
}
tst.incTotalMergeBackoffSeconds(time.Since(sleepStart).Seconds())
}
- dispatch, maxPartID := tst.mergeControl.beginDispatch()
- if !dispatch {
- continue
- }
- if tst.dispatchAllMergesUpTo(threshold, fastCh, slowCh, maxPartID) {
- tst.mergeControl.endDispatch()
+ if tst.runDispatchCycle(threshold, fastCh, slowCh) {
return
}
- tst.mergeControl.endDispatch()
}
}
}
+// runDispatchCycle executes one selection/dispatch cycle, converting panics into a
+// logged, recovered failure so a selection-time panic cannot kill the dispatcher.
+func (tst *tsTable) runDispatchCycle(threshold uint64, fastCh, slowCh chan *mergeDispatchRequest) (stop bool) {
+ dispatch, maxPartID := tst.mergeControl.beginDispatch()
+ if !dispatch {
+ return false
+ }
+ defer tst.mergeControl.endDispatch()
+ defer func() {
+ if panicVal := recover(); panicVal != nil {
+ tst.incTotalMergePanicRecovered(1)
+ tst.l.Error().Str("panic", fmt.Sprintf("%v", panicVal)).Str("stack", string(debug.Stack())).Msg("merge dispatch cycle panicked")
+ }
+ }()
+ return tst.dispatchAllMergesUpTo(threshold, fastCh, slowCh, maxPartID)
+}
+
func (tst *tsTable) dispatchAllMerges(threshold uint64, fastCh, slowCh chan *mergeDispatchRequest) bool {
return tst.dispatchAllMergesUpTo(threshold, fastCh, slowCh, 0)
}
@@ -486,10 +498,19 @@ func (tst *tsTable) mergeLaneWorker(ch chan *mergeDispatchRequest, merges chan *
}
tst.incTotalMergeLoopStarted(1)
- _, mergeErr := tst.mergePartsThenSendIntroductionObserved(
- snapshotCreatorMerger, req.parts, req.toBeMerged, merges,
- tst.loopCloser.CloseNotify(), req.typ, req.lane, nil, req.benchmark,
- )
+ mergeErr := func() (attemptErr error) {
+ defer func() {
+ if panicVal := recover(); panicVal != nil {
+ tst.incTotalMergePanicRecovered(1)
+ attemptErr = fmt.Errorf("merge panicked: %v\n%s", panicVal, debug.Stack())
+ }
+ }()
+ _, attemptErr = tst.mergePartsThenSendIntroductionObserved(
+ snapshotCreatorMerger, req.parts, req.toBeMerged, merges,
+ tst.loopCloser.CloseNotify(), req.typ, req.lane, nil, req.benchmark,
+ )
+ return attemptErr
+ }()
tst.incTotalMergeLoopFinished(1)
if attributionAcquired {
<-tst.attributionCh
diff --git a/banyand/trace/metrics.go b/banyand/trace/metrics.go
index 90a2478f6..b64e58ffe 100644
--- a/banyand/trace/metrics.go
+++ b/banyand/trace/metrics.go
@@ -68,6 +68,7 @@ type metrics struct {
totalMergeQueueLatency meter.Counter
totalMergePartQuarantined meter.Counter
totalMergeBackoffSeconds meter.Counter
+ totalMergePanicRecovered meter.Counter
pipelineTracesEvaluated meter.Counter
pipelineTracesDropped meter.Counter
@@ -293,6 +294,15 @@ func (tst *tsTable) incTotalMergeBackoffSeconds(delta float64) {
tst.metrics.totalMergeBackoffSeconds.Inc(delta)
}
+// incTotalMergePanicRecovered counts a panic caught and converted into an ordinary merge
+// failure by the Fix D backstop (worker or dispatcher).
+func (tst *tsTable) incTotalMergePanicRecovered(delta int) {
+ if tst == nil || tst.metrics == nil {
+ return
+ }
+ tst.metrics.totalMergePanicRecovered.Inc(float64(delta))
+}
+
// The pipeline metric increment helpers below are wired into the merge filter by
// the config-driven activation story.
@@ -463,6 +473,7 @@ func (m *metrics) DeleteAll() {
m.totalMergeQueueLatency.Delete("file", "slow")
m.totalMergePartQuarantined.Delete()
m.totalMergeBackoffSeconds.Delete()
+ m.totalMergePanicRecovered.Delete()
m.pipelineTracesEvaluated.Delete()
m.pipelineTracesDropped.Delete()
@@ -511,6 +522,7 @@ func (s *supplier) newMetrics(p common.Position) storage.Metrics {
totalMergeQueueLatency: factory.NewCounter("total_merge_queue_latency", "type", "lane"),
totalMergePartQuarantined: factory.NewCounter("total_merge_part_quarantined"),
totalMergeBackoffSeconds: factory.NewCounter("total_merge_backoff_seconds"),
+ totalMergePanicRecovered: factory.NewCounter("total_merge_panic_recovered"),
pipelineTracesEvaluated: factory.NewCounter("pipeline_traces_evaluated"),
pipelineTracesDropped: factory.NewCounter("pipeline_traces_dropped"),
pipelineTracesRetained: factory.NewCounter("pipeline_traces_retained"),
@@ -575,6 +587,7 @@ func (qs *queueSupplier) newMetrics(p common.Position) (storage.Metrics, observa
totalMergeQueueLatency: factory.NewCounter("total_merge_queue_latency", "type", "lane"),
totalMergePartQuarantined: factory.NewCounter("total_merge_part_quarantined"),
totalMergeBackoffSeconds: factory.NewCounter("total_merge_backoff_seconds"),
+ totalMergePanicRecovered: factory.NewCounter("total_merge_panic_recovered"),
pipelineTracesEvaluated: factory.NewCounter("pipeline_traces_evaluated"),
pipelineTracesDropped: factory.NewCounter("pipeline_traces_dropped"),
pipelineTracesRetained: factory.NewCounter("pipeline_traces_retained"),
(skywalking-banyandb) 06/06: docs: record the trace merge failure resilience fixes in CHANGES
Posted by ha...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
hanahmily pushed a commit to branch trace-merge-resilience
in repository https://gitbox.apache.org/repos/asf/skywalking-banyandb.git
commit 20c2a2edde55d58015e8937aa9382bc64caf27f1
Author: owenwillison-vajra <30...@qq.com>
AuthorDate: Sat Aug 8 23:51:49 2026 +0000
docs: record the trace merge failure resilience fixes in CHANGES
---
CHANGES.md | 1 +
1 file changed, 1 insertion(+)
diff --git a/CHANGES.md b/CHANGES.md
index 7fa63a795..b0c8ceae3 100644
--- a/CHANGES.md
+++ b/CHANGES.md
@@ -154,6 +154,7 @@ Release Notes.
- Allow `*` as a non-initial character in BydbQL identifiers, so resources named with `*` become queryable, not just writable.
- Enable periodic health checks on the queue client (`--<prefix>-client-health-check-interval`, default `10s`), evicting dead data nodes proactively.
- Build the plugin `.so` with `-tags slim` to match the host binary's ABI. Without this, `pkg/pool` has build-tag-dependent code (`tracker.go` vs `tracker_stub.go`) and the plugin-sidecar E2E failed with "plugin was built with a different version of package `pkg/pool`".
+- Make the trace merge loop survive a persistently failing merge instead of destroying the node. A production cluster wedged when a few unreadable parts made every merge selection fail: each failed attempt leaked its partially-written output directory (the block writer creates it before the merge runs), which exhausted the volume's inodes at 8% byte usage; the resulting disk-full panic then killed the merge lane workers, which are recovered but never respawned, so the dispatcher blocked [...]
### Document
(skywalking-banyandb) 03/06: feat(trace): back off merge dispatch after consecutive failures
Posted by ha...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
hanahmily pushed a commit to branch trace-merge-resilience
in repository https://gitbox.apache.org/repos/asf/skywalking-banyandb.git
commit 84c6aba875f6cdafbf43f11c29aa1a8398c8d2c3
Author: owenwillison-vajra <30...@qq.com>
AuthorDate: Sat Aug 8 13:45:27 2026 +0000
feat(trace): back off merge dispatch after consecutive failures
A persistently failing merge was retried at flush/sync notification
cadence (~4/sec) forever. Track consecutive merge failures in
mergeLoopControl and delay the next dispatch exponentially from 1s up
to a 60s cap, reset on any success. The dispatcher sleeps before
beginDispatch so wave waiters are not delayed, wakes early on any
control-state change, and wave mode bypasses backoff entirely so
operator-driven controlled merges stay prompt. New metric:
total_merge_backoff_seconds.
---
banyand/trace/merge_backoff_test.go | 98 ++++++++++++++++++++++++++++++++++
banyand/trace/merge_control.go | 102 +++++++++++++++++++++++++++++-------
banyand/trace/merger.go | 21 ++++++++
banyand/trace/metrics.go | 13 +++++
4 files changed, 216 insertions(+), 18 deletions(-)
diff --git a/banyand/trace/merge_backoff_test.go b/banyand/trace/merge_backoff_test.go
new file mode 100644
index 000000000..a13985795
--- /dev/null
+++ b/banyand/trace/merge_backoff_test.go
@@ -0,0 +1,98 @@
+// Licensed to the Apache Software Foundation (ASF) under one or more contributor
+// license agreements. See the NOTICE file distributed with this work for
+// additional information regarding copyright ownership. The ASF licenses this
+// file to you under the Apache License, Version 2.0 (the "License"); you may not
+// use this file except in compliance with the License. You may obtain a copy of
+// the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
+// WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
+// License for the specific language governing permissions and limitations
+// under the License.
+
+package trace
+
+import (
+ "testing"
+ "time"
+
+ "github.com/stretchr/testify/require"
+)
+
+// TestRecordOutcomeBackoffProgression asserts the exponential schedule doubles from
+// mergeBackoffBase up to mergeBackoffCap, and that a success clears it immediately.
+func TestRecordOutcomeBackoffProgression(t *testing.T) {
+ mc := newMergeLoopControl()
+ fixedNow := time.Date(2026, 8, 8, 0, 0, 0, 0, time.UTC)
+ mc.nowFunc = func() time.Time { return fixedNow }
+
+ expected := []time.Duration{
+ time.Second, 2 * time.Second, 4 * time.Second, 8 * time.Second,
+ 16 * time.Second, 32 * time.Second, 60 * time.Second, 60 * time.Second,
+ 60 * time.Second, 60 * time.Second,
+ }
+ for i, want := range expected {
+ mc.recordOutcome(false)
+ got := mc.backoffUntil.Sub(fixedNow)
+ require.Equalf(t, want, got, "failure %d: expected backoff delay %s, got %s", i+1, want, got)
+ }
+ require.Equal(t, len(expected), mc.consecutiveFailures)
+
+ remaining := mc.backoffRemaining()
+ require.Greater(t, remaining, time.Duration(0), "backoff should still be active before a success is recorded")
+
+ mc.recordOutcome(true)
+ require.Zero(t, mc.consecutiveFailures)
+ require.True(t, mc.backoffUntil.IsZero())
+ require.Zero(t, mc.backoffRemaining())
+}
+
+// TestBackoffWaveBypass asserts wave mode bypasses an active backoff, and that the
+// backoff reasserts itself once the wave releases.
+func TestBackoffWaveBypass(t *testing.T) {
+ mc := newMergeLoopControl()
+ fixedNow := time.Date(2026, 8, 8, 0, 0, 0, 0, time.UTC)
+ mc.nowFunc = func() time.Time { return fixedNow }
+
+ mc.recordOutcome(false)
+ mc.recordOutcome(false)
+ require.Greater(t, mc.backoffRemaining(), time.Duration(0))
+
+ mc.startWave(0, 0)
+ require.Zero(t, mc.backoffRemaining(), "wave mode must bypass an active backoff")
+
+ mc.releaseWave()
+ require.Greater(t, mc.backoffRemaining(), time.Duration(0), "backoff resumes once the wave releases")
+}
+
+// TestDispatcherHonorsBackoff drives mergeControl through two failing dispatch cycles the
+// way dispatcherLoop and mergeLaneWorker do (beginDispatch/endDispatch bracketing a
+// recordOutcome(false)) and asserts the control state — rather than wall-clock timing —
+// reflects an active, growing backoff that a subsequent trigger must honor.
+func TestDispatcherHonorsBackoff(t *testing.T) {
+ mc := newMergeLoopControl()
+ fixedNow := time.Date(2026, 8, 8, 0, 0, 0, 0, time.UTC)
+ mc.nowFunc = func() time.Time { return fixedNow }
+
+ require.NoError(t, mc.enqueue(nil))
+ <-mc.trigger
+ dispatch, _ := mc.beginDispatch()
+ require.True(t, dispatch)
+ mc.recordOutcome(false)
+ mc.endDispatch()
+ require.Equal(t, 1, mc.consecutiveFailures)
+ require.Equal(t, time.Second, mc.backoffUntil.Sub(fixedNow))
+
+ require.NoError(t, mc.enqueue(nil))
+ <-mc.trigger
+ dispatch, _ = mc.beginDispatch()
+ require.True(t, dispatch)
+ mc.recordOutcome(false)
+ mc.endDispatch()
+ require.Equal(t, 2, mc.consecutiveFailures)
+ require.Equal(t, 2*time.Second, mc.backoffUntil.Sub(fixedNow))
+ require.Greater(t, mc.backoffRemaining(), time.Duration(0))
+}
diff --git a/banyand/trace/merge_control.go b/banyand/trace/merge_control.go
index 01210f36e..ced1c6a85 100644
--- a/banyand/trace/merge_control.go
+++ b/banyand/trace/merge_control.go
@@ -21,29 +21,43 @@ import (
"fmt"
"sync"
"sync/atomic"
+ "time"
)
var errMergeLoopUnavailable = errors.New("merge loop is unavailable")
+// mergeBackoffBase is the initial backoff delay applied after the first consecutive
+// merge failure; mergeBackoffCap is the maximum delay the exponential schedule reaches.
+// mergeBackoffMaxShift bounds the doubling exponent so the shift never overflows;
+// consecutive failures beyond it are clamped directly to mergeBackoffCap.
+const (
+ mergeBackoffBase = time.Second
+ mergeBackoffCap = 60 * time.Second
+ mergeBackoffMaxShift = 7
+)
+
type mergeLoopControl struct {
- changed chan struct{}
- trigger chan struct{}
- version uint64
- workVersion uint64
- emptyEpoch uint64
- emptyWorkVersion uint64
- waveMaxPartID uint64
- queued int
- running int
- triggerPending bool
- dispatcherActive bool
- emptyEpochObserved bool
- stopped bool
- waveMode bool
- waveBlocked bool
- waveDispatchLimit int
- waveDispatches int
- mu sync.Mutex
+ changed chan struct{}
+ trigger chan struct{}
+ nowFunc func() time.Time
+ version uint64
+ workVersion uint64
+ emptyEpoch uint64
+ emptyWorkVersion uint64
+ waveMaxPartID uint64
+ backoffUntil time.Time
+ queued int
+ running int
+ consecutiveFailures int
+ triggerPending bool
+ dispatcherActive bool
+ emptyEpochObserved bool
+ stopped bool
+ waveMode bool
+ waveBlocked bool
+ waveDispatchLimit int
+ waveDispatches int
+ mu sync.Mutex
}
type mergeLoopState struct {
@@ -75,6 +89,58 @@ func (mc *mergeLoopControl) notifyLocked() {
mc.changed = make(chan struct{})
}
+// now returns nowFunc's value, or time.Now if nowFunc is unset; a testability seam for
+// deterministic backoff assertions.
+func (mc *mergeLoopControl) now() time.Time {
+ if mc.nowFunc != nil {
+ return mc.nowFunc()
+ }
+ return time.Now()
+}
+
+// recordOutcome reports the result of one merge attempt. A success clears the backoff
+// state; a failure extends it exponentially, doubling from mergeBackoffBase up to
+// mergeBackoffCap.
+func (mc *mergeLoopControl) recordOutcome(success bool) {
+ mc.mu.Lock()
+ defer mc.mu.Unlock()
+ if success {
+ mc.consecutiveFailures = 0
+ mc.backoffUntil = time.Time{}
+ mc.notifyLocked()
+ return
+ }
+ mc.consecutiveFailures++
+ delay := mergeBackoffCap
+ if mc.consecutiveFailures <= mergeBackoffMaxShift {
+ delay = mergeBackoffBase << uint(mc.consecutiveFailures-1)
+ if delay > mergeBackoffCap {
+ delay = mergeBackoffCap
+ }
+ }
+ mc.backoffUntil = mc.now().Add(delay)
+ mc.notifyLocked()
+}
+
+// backoffRemaining reports how long the dispatcher should still sleep before dispatching
+// another selection cycle. Wave mode always bypasses backoff: waves are operator-driven
+// and must fail visibly and immediately.
+func (mc *mergeLoopControl) backoffRemaining() time.Duration {
+ mc.mu.Lock()
+ defer mc.mu.Unlock()
+ if mc.waveMode {
+ return 0
+ }
+ if mc.backoffUntil.IsZero() {
+ return 0
+ }
+ now := mc.now()
+ if !mc.backoffUntil.After(now) {
+ return 0
+ }
+ return mc.backoffUntil.Sub(now)
+}
+
func (mc *mergeLoopControl) enqueue(closeCh <-chan struct{}) error {
mc.mu.Lock()
if mc.stopped {
diff --git a/banyand/trace/merger.go b/banyand/trace/merger.go
index 9c81522eb..c97a2a858 100644
--- a/banyand/trace/merger.go
+++ b/banyand/trace/merger.go
@@ -325,6 +325,22 @@ func (tst *tsTable) dispatcherLoop(threshold uint64, fastCh, slowCh chan *mergeD
case <-tst.loopCloser.CloseNotify():
return
case <-tst.mergeControl.trigger:
+ for {
+ delay := tst.mergeControl.backoffRemaining()
+ if delay <= 0 {
+ break
+ }
+ state := tst.mergeControl.state()
+ sleepStart := time.Now()
+ select {
+ case <-time.After(delay):
+ case <-state.changed:
+ case <-tst.loopCloser.CloseNotify():
+ tst.incTotalMergeBackoffSeconds(time.Since(sleepStart).Seconds())
+ return
+ }
+ tst.incTotalMergeBackoffSeconds(time.Since(sleepStart).Seconds())
+ }
dispatch, maxPartID := tst.mergeControl.beginDispatch()
if !dispatch {
continue
@@ -490,7 +506,12 @@ func (tst *tsTable) mergeLaneWorker(ch chan *mergeDispatchRequest, merges chan *
tst.l.Logger.Warn().Err(mergeErr).Str("typ", req.typ).Str("lane", req.lane).Msg("merge lane worker error")
tst.incTotalMergeLoopErr(1)
tst.recordUnreadablePart(mergeErr)
+ if tst.mergeControl != nil {
+ tst.mergeControl.recordOutcome(false)
+ }
}
+ } else if tst.mergeControl != nil {
+ tst.mergeControl.recordOutcome(true)
}
}
}
diff --git a/banyand/trace/metrics.go b/banyand/trace/metrics.go
index 0a7157ece..90a2478f6 100644
--- a/banyand/trace/metrics.go
+++ b/banyand/trace/metrics.go
@@ -67,6 +67,7 @@ type metrics struct {
totalMerged meter.Counter
totalMergeQueueLatency meter.Counter
totalMergePartQuarantined meter.Counter
+ totalMergeBackoffSeconds meter.Counter
pipelineTracesEvaluated meter.Counter
pipelineTracesDropped meter.Counter
@@ -283,6 +284,15 @@ func (tst *tsTable) incTotalMergePartQuarantined(delta int) {
tst.metrics.totalMergePartQuarantined.Inc(float64(delta))
}
+// incTotalMergeBackoffSeconds accumulates the time the dispatcher spent sleeping due to
+// repeated merge failures (Fix C).
+func (tst *tsTable) incTotalMergeBackoffSeconds(delta float64) {
+ if tst == nil || tst.metrics == nil {
+ return
+ }
+ tst.metrics.totalMergeBackoffSeconds.Inc(delta)
+}
+
// The pipeline metric increment helpers below are wired into the merge filter by
// the config-driven activation story.
@@ -452,6 +462,7 @@ func (m *metrics) DeleteAll() {
m.totalMergeQueueLatency.Delete("file", "fast")
m.totalMergeQueueLatency.Delete("file", "slow")
m.totalMergePartQuarantined.Delete()
+ m.totalMergeBackoffSeconds.Delete()
m.pipelineTracesEvaluated.Delete()
m.pipelineTracesDropped.Delete()
@@ -499,6 +510,7 @@ func (s *supplier) newMetrics(p common.Position) storage.Metrics {
totalMerged: factory.NewCounter("total_merged", "type", "lane"),
totalMergeQueueLatency: factory.NewCounter("total_merge_queue_latency", "type", "lane"),
totalMergePartQuarantined: factory.NewCounter("total_merge_part_quarantined"),
+ totalMergeBackoffSeconds: factory.NewCounter("total_merge_backoff_seconds"),
pipelineTracesEvaluated: factory.NewCounter("pipeline_traces_evaluated"),
pipelineTracesDropped: factory.NewCounter("pipeline_traces_dropped"),
pipelineTracesRetained: factory.NewCounter("pipeline_traces_retained"),
@@ -562,6 +574,7 @@ func (qs *queueSupplier) newMetrics(p common.Position) (storage.Metrics, observa
totalMerged: factory.NewCounter("total_merged", "type", "lane"),
totalMergeQueueLatency: factory.NewCounter("total_merge_queue_latency", "type", "lane"),
totalMergePartQuarantined: factory.NewCounter("total_merge_part_quarantined"),
+ totalMergeBackoffSeconds: factory.NewCounter("total_merge_backoff_seconds"),
pipelineTracesEvaluated: factory.NewCounter("pipeline_traces_evaluated"),
pipelineTracesDropped: factory.NewCounter("pipeline_traces_dropped"),
pipelineTracesRetained: factory.NewCounter("pipeline_traces_retained"),