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 6f7d712f4 fix(storage): prevent DiskMonitor.Stop from hanging when
forced cleanup is active, leaking the whole server on shutdown (#1216)
6f7d712f4 is described below
commit 6f7d712f4f78445b531a4cdbd76efe4a51f7e0c5
Author: Gao Hongtao <[email protected]>
AuthorDate: Mon Jul 13 15:52:25 2026 +0800
fix(storage): prevent DiskMonitor.Stop from hanging when forced cleanup is
active, leaking the whole server on shutdown (#1216)
---
banyand/internal/storage/disk_monitor.go | 42 +++++++++++++++---
banyand/internal/storage/disk_monitor_test.go | 63 +++++++++++++++++++++++++++
banyand/measure/metadata.go | 1 +
banyand/measure/metadata_test.go | 36 ++++++++++++++-
4 files changed, 135 insertions(+), 7 deletions(-)
diff --git a/banyand/internal/storage/disk_monitor.go
b/banyand/internal/storage/disk_monitor.go
index 58a8debde..7a3b8abc3 100644
--- a/banyand/internal/storage/disk_monitor.go
+++ b/banyand/internal/storage/disk_monitor.go
@@ -75,10 +75,12 @@ type DiskMonitor struct {
logger *logger.Logger
ticker *time.Ticker
stopCh chan struct{}
+ doneCh chan struct{}
metrics *diskMonitorMetrics
config RetentionConfig
snapshotMaxAge time.Duration
isActive atomic.Bool
+ started atomic.Bool
}
type diskMonitorMetrics struct {
@@ -136,6 +138,7 @@ func NewDiskMonitor(service RetentionService, config
RetentionConfig, omr observ
config: config,
logger: logger,
stopCh: make(chan struct{}),
+ doneCh: make(chan struct{}),
metrics: metrics,
snapshotMaxAge: 24 * time.Hour, // Always keep snapshots newer
than 24h
}
@@ -143,8 +146,14 @@ func NewDiskMonitor(service RetentionService, config
RetentionConfig, omr observ
// Start begins monitoring disk usage and starts the forced retention process.
func (dm *DiskMonitor) Start() {
+ // Mark the monitor as started so Stop() knows a doneCh close is
guaranteed
+ // (either from the disabled path below or from monitorLoop on exit).
+ dm.started.Store(true)
+
if dm.config.CheckInterval <= 0 {
dm.logger.Warn().Msg("disk monitor check interval is 0 or
negative, monitor disabled")
+ // No monitor loop will run, so release Stop()'s wait
immediately.
+ close(dm.doneCh)
return
}
@@ -164,20 +173,37 @@ func (dm *DiskMonitor) Start() {
// Stop stops the disk monitor gracefully.
func (dm *DiskMonitor) Stop() {
+ // If Start() was never called, no monitor loop exists and doneCh will
never
+ // be closed, so there is nothing to stop or wait for. Returning here
keeps
+ // Stop() safe to call on a constructed-but-unstarted monitor.
+ if !dm.started.Load() {
+ dm.logger.Info().Msg("disk monitor stopped")
+ return
+ }
+
if dm.ticker != nil {
dm.ticker.Stop()
}
close(dm.stopCh)
- // Wait for any active cleanup to finish
- for dm.isActive.Load() {
- time.Sleep(100 * time.Millisecond)
- }
+ // Wait for the monitor loop (including any in-flight cleanup cycle) to
+ // exit. The loop clears isActive on exit, so we must not poll isActive
+ // here: once stopCh is closed the loop may return without ever running
+ // another cleanup cycle, leaving isActive set and hanging Stop()
forever.
+ <-dm.doneCh
dm.logger.Info().Msg("disk monitor stopped")
}
func (dm *DiskMonitor) monitorLoop(serviceName string) {
+ defer func() {
+ // Deactivate on exit so a forced cleanup that is still active
when the
+ // monitor stops does not leave the monitor marked active, and
to unblock
+ // Stop() which waits on doneCh.
+ dm.isActive.Store(false)
+ dm.metrics.forcedRetentionActive.Set(0, serviceName)
+ close(dm.doneCh)
+ }()
for {
select {
case <-dm.stopCh:
@@ -296,9 +322,13 @@ func (dm *DiskMonitor) runForcedCleanup(serviceName
string, _ int) {
return
}
- // Sleep for cooldown period
+ // Sleep for cooldown period, aborting early if the monitor is
stopping
+ // so shutdown is not delayed by a full cooldown interval.
dm.logger.Debug().Dur("cooldown",
dm.config.Cooldown).Msg("cooling down between deletions")
- time.Sleep(dm.config.Cooldown)
+ select {
+ case <-time.After(dm.config.Cooldown):
+ case <-dm.stopCh:
+ }
} else {
// No more segments to delete, stop cleanup
dm.logger.Warn().Msg("no more segments available for deletion,
stopping forced cleanup")
diff --git a/banyand/internal/storage/disk_monitor_test.go
b/banyand/internal/storage/disk_monitor_test.go
index aad31c3d2..1d8418cf2 100644
--- a/banyand/internal/storage/disk_monitor_test.go
+++ b/banyand/internal/storage/disk_monitor_test.go
@@ -244,6 +244,69 @@ func TestDiskMonitor_StartStop(t *testing.T) {
assert.False(t, dm.isActive.Load())
}
+func TestDiskMonitor_StopDoesNotHangWhenActive(t *testing.T) {
+ service := NewMockRetentionService()
+ config := RetentionConfig{
+ HighWatermark: 80.0,
+ LowWatermark: 60.0,
+ CheckInterval: time.Hour, // long interval so no cleanup
cycle races the test
+ Cooldown: time.Millisecond,
+ ForceCleanupEnabled: true,
+ }
+ registry := createMockMetricsRegistry()
+
+ dm := NewDiskMonitor(service, config, registry)
+ dm.Start()
+
+ // Simulate a forced cleanup that is still active when the monitor is
+ // stopped. Previously Stop() polled isActive, which the monitor loop
only
+ // clears from within a cleanup cycle; closing stopCh made the loop exit
+ // without clearing it, hanging Stop() forever.
+ dm.isActive.Store(true)
+
+ stopped := make(chan struct{})
+ go func() {
+ dm.Stop()
+ close(stopped)
+ }()
+
+ select {
+ case <-stopped:
+ case <-time.After(5 * time.Second):
+ t.Fatal("Stop() hung while a forced cleanup was active")
+ }
+ assert.False(t, dm.isActive.Load())
+}
+
+func TestDiskMonitor_StopBeforeStartDoesNotHang(t *testing.T) {
+ service := NewMockRetentionService()
+ config := RetentionConfig{
+ HighWatermark: 80.0,
+ LowWatermark: 60.0,
+ CheckInterval: time.Hour,
+ Cooldown: time.Millisecond,
+ ForceCleanupEnabled: true,
+ }
+ registry := createMockMetricsRegistry()
+
+ dm := NewDiskMonitor(service, config, registry)
+
+ // Stop() must be safe on a constructed-but-unstarted monitor. Stop()
waits
+ // on doneCh, which only Start()/monitorLoop close, so calling it before
+ // Start() previously blocked forever.
+ stopped := make(chan struct{})
+ go func() {
+ dm.Stop()
+ close(stopped)
+ }()
+
+ select {
+ case <-stopped:
+ case <-time.After(5 * time.Second):
+ t.Fatal("Stop() hung when called before Start()")
+ }
+}
+
func TestDiskMonitor_StartWithZeroInterval(t *testing.T) {
service := NewMockRetentionService()
config := RetentionConfig{
diff --git a/banyand/measure/metadata.go b/banyand/measure/metadata.go
index 9db478549..550c697c6 100644
--- a/banyand/measure/metadata.go
+++ b/banyand/measure/metadata.go
@@ -524,6 +524,7 @@ func (sr *schemaRepo) collectShardInfo(table any, shardID
uint32) *databasev1.Sh
PartCount: int64(partCount),
InvertedIndexInfo: &databasev1.InvertedIndexInfo{},
SidxInfo: &databasev1.SIDXInfo{},
+ FilePartCount: int64(filePartCount),
}
}
diff --git a/banyand/measure/metadata_test.go b/banyand/measure/metadata_test.go
index ef30f0c1a..a9d9a6baf 100644
--- a/banyand/measure/metadata_test.go
+++ b/banyand/measure/metadata_test.go
@@ -512,16 +512,33 @@ var _ = Describe("Schema Change", func() {
env := setupSchemaChangeMeasure(svcs, measureName,
measureSetupOptions{withExtraTag: true})
writeSchemaChangeMeasureData(svcs, measureName,
now.Add(-2*time.Hour), 5,
measureWriteDataOptions{withExtraTag: true})
+ filePartCountAfterFirstBatch, filePartCountErr :=
getMeasureFilePartCount(svcs, groupName)
+ Expect(filePartCountErr).ShouldNot(HaveOccurred())
changeExtraMeasureTagType(svcs, measureName)
writeSchemaChangeMeasureData(svcs, measureName,
now.Add(-1*time.Hour), 3,
measureWriteDataOptions{withExtraTagString:
true, entityIDPrefix: "entity_new_"})
+ // Wait for the second batch to flush to disk, creating
additional
+ // file parts that the merge loop can pick up. Without
this gate the
+ // merge Eventually below also has to absorb flush
latency, which can
+ // exceed 30 s on resource-constrained CI runners with
-race.
+ Eventually(func(innerGm Gomega) int64 {
+ currentFilePartCount, currentFilePartCountErr
:= getMeasureFilePartCount(svcs, groupName)
+
innerGm.Expect(currentFilePartCountErr).ShouldNot(HaveOccurred())
+ return currentFilePartCount
+ }, flags.EventuallyTimeout).Should(BeNumerically(">",
filePartCountAfterFirstBatch))
partCountBeforeMerge, partCountErr :=
getTotalMeasurePartCount(svcs, groupName)
Expect(partCountErr).ShouldNot(HaveOccurred())
+ // The background merge runs asynchronously; under the
full parallel -race
+ // suite on CI its goroutine competes for CPU and
snapshot locks with every
+ // other test. Poll on a relaxed interval (rather than
Gomega's ~10ms
+ // default) so this wait does not starve the merge it
is waiting for, with a
+ // generous, environment-scaled budget for merge
latency. The post-merge
+ // part count is stable (no further writes), so slow
polling never misses it.
Eventually(func(innerGm Gomega) int64 {
currentPartCount, currentPartCountErr :=
getTotalMeasurePartCount(svcs, groupName)
innerGm.Expect(currentPartCountErr).ShouldNot(HaveOccurred())
return currentPartCount
- }, flags.EventuallyTimeout).Should(BeNumerically("<",
partCountBeforeMerge))
+ }, 10*flags.EventuallyTimeout,
500*time.Millisecond).Should(BeNumerically("<", partCountBeforeMerge))
Eventually(func(innerGm Gomega) {
dataPoints :=
querySchemaChangeMeasureData(svcs, measureName,
@@ -1108,3 +1125,20 @@ func getTotalMeasurePartCount(svcs *services, group
string) (int64, error) {
}
return total, nil
}
+
+func getMeasureFilePartCount(svcs *services, group string) (int64, error) {
+ dataInfo, err := svcs.measure.CollectDataInfo(context.TODO(), group)
+ if err != nil {
+ return 0, fmt.Errorf("collect measure data info: %w", err)
+ }
+ if dataInfo == nil {
+ return 0, errors.New("measure data info is nil")
+ }
+ var total int64
+ for _, seg := range dataInfo.SegmentInfo {
+ for _, shard := range seg.ShardInfo {
+ total += shard.FilePartCount
+ }
+ }
+ return total, nil
+}
