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 ac7f7a3b1 feat(lifecycle): throttle migration under receiver memory
pressure; fix bluge dedup OOM (#1193)
ac7f7a3b1 is described below
commit ac7f7a3b191e0aa2be919dcea311a03f34770a99
Author: mrproliu <[email protected]>
AuthorDate: Sat Jun 27 08:16:44 2026 +0800
feat(lifecycle): throttle migration under receiver memory pressure; fix
bluge dedup OOM (#1193)
* feat(lifecycle): throttle migration under receiver memory pressure; fix
bluge dedup OOM
---
CHANGES.md | 2 +
api/proto/banyandb/cluster/v1/rpc.proto | 1 +
.../backup/lifecycle/measure_migration_visitor.go | 72 ++--
banyand/backup/lifecycle/migration_error_test.go | 7 +-
banyand/backup/lifecycle/migration_integration.go | 13 +-
banyand/backup/lifecycle/retry.go | 135 +++++++
banyand/backup/lifecycle/retry_test.go | 132 +++++++
banyand/backup/lifecycle/row_replay_measure.go | 6 +-
banyand/backup/lifecycle/row_replay_pipeline.go | 4 +-
banyand/backup/lifecycle/row_replay_stream.go | 2 +-
banyand/backup/lifecycle/row_replay_trace.go | 2 +-
banyand/backup/lifecycle/service.go | 31 +-
.../backup/lifecycle/stream_migration_visitor.go | 110 +++---
.../backup/lifecycle/trace_migration_visitor.go | 135 ++++---
banyand/measure/export_test.go | 37 ++
banyand/measure/measure.go | 1 +
banyand/measure/migration_oom_throttling_test.go | 408 +++++++++++++++++++++
banyand/measure/svc_data.go | 4 +-
banyand/measure/svc_liaison.go | 3 +
banyand/measure/svc_standalone.go | 2 +
banyand/measure/write_data.go | 46 ++-
banyand/measure/write_data_busy_test.go | 67 ++++
banyand/protector/wait.go | 63 ++++
banyand/protector/wait_test.go | 72 ++++
banyand/queue/pub/chunked_sync.go | 6 +
banyand/queue/queue.go | 18 +-
banyand/queue/queue_test.go | 35 ++
banyand/queue/sub/chunked_sync.go | 5 +
banyand/queue/sub/chunked_sync_test.go | 62 ++++
banyand/stream/stream.go | 1 +
banyand/stream/svc_liaison.go | 3 +
banyand/stream/svc_standalone.go | 6 +-
banyand/stream/write_data.go | 91 +++--
banyand/stream/write_data_busy_test.go | 94 +++++
banyand/trace/svc_liaison.go | 2 +
banyand/trace/svc_standalone.go | 4 +-
banyand/trace/trace.go | 1 +
banyand/trace/write_data.go | 47 ++-
banyand/trace/write_data_busy_test.go | 67 ++++
dist/LICENSE | 2 +-
docs/api-reference.md | 1 +
docs/operation/configuration.md | 3 +
docs/operation/lifecycle.md | 1 +
go.mod | 2 +-
go.sum | 4 +-
45 files changed, 1601 insertions(+), 209 deletions(-)
diff --git a/CHANGES.md b/CHANGES.md
index cb5c81956..21a197d0b 100644
--- a/CHANGES.md
+++ b/CHANGES.md
@@ -68,6 +68,8 @@ Release Notes.
- Enhance segment lifecycle: `refCount` now counts only active users,
decoupled from "open" (`index != nil`), adding a "dormant" state (open,
`refCount == 0`). A `DecRef` to zero no longer closes a segment; idle reclaim
and retention delete act only at `refCount == 0`, so an in-flight
snapshot/inspect is no longer torn down mid-operation (fixing the cold-node
nil-index panic and bluge lock churn) while idle segments still release their
bluge writers.
- Speed up GCS backup uploads: write each object and its checksum metadata in
one request, dropping the per-object `Update` round-trip.
- Lifecycle migration now archives rows whose measure/stream schema was
deleted from the registry, instead of aborting the group.
+- Throttle lifecycle migration under receiver memory pressure so it no longer
OOM-kills the destination data node, via three always-on mechanisms: per-chunk
`SYNC_STATUS_SERVER_BUSY` load-shed (the sender backs off and retries); a
bounded memory wait before introducing an external series-index segment
(`{measure,stream,trace}-lifecycle-receive-mem-wait-timeout`, default 5m); and
sender-side bounded-backoff retry of transient send failures — target node
restarting/`no nodes available`, di [...]
+- Cut the lifecycle external series-index deduplicator's memory from ~1.8 GB
to near-zero per segment by bumping bluge to a memory-efficient implementation:
duplicates are now resolved via `_id` term-dictionary point lookups
(`Dictionary.Contains`) instead of decompressing and retaining every segment's
stored fields, removing the dominant heap source during external-segment
introduction. No public API change and identical dedup semantics (verified by a
differential test); this is the roo [...]
### Bug Fixes
diff --git a/api/proto/banyandb/cluster/v1/rpc.proto
b/api/proto/banyandb/cluster/v1/rpc.proto
index aec1b9963..a763acb8d 100644
--- a/api/proto/banyandb/cluster/v1/rpc.proto
+++ b/api/proto/banyandb/cluster/v1/rpc.proto
@@ -182,6 +182,7 @@ enum SyncStatus {
SYNC_STATUS_SYNC_COMPLETE = 5; // Entire sync operation completed
successfully.
SYNC_STATUS_VERSION_UNSUPPORTED = 6; // Version not supported for sync
operations.
SYNC_STATUS_FORMAT_VERSION_MISMATCH = 7; // File format version incompatible.
+ SYNC_STATUS_SERVER_BUSY = 8; // Receiver under memory pressure; sender
should back off and retry the whole part.
}
service Service {
diff --git a/banyand/backup/lifecycle/measure_migration_visitor.go
b/banyand/backup/lifecycle/measure_migration_visitor.go
index 2059922b9..b63493e80 100644
--- a/banyand/backup/lifecycle/measure_migration_visitor.go
+++ b/banyand/backup/lifecycle/measure_migration_visitor.go
@@ -41,6 +41,7 @@ import (
// measureMigrationVisitor implements the measure.Visitor interface for
file-based migration.
type measureMigrationVisitor struct {
+ ctx context.Context
client queue.Client
lfs fs.FileSystem
metadata metadata.Repo
@@ -64,11 +65,12 @@ type measureMigrationVisitor struct {
}
// newMeasureMigrationVisitor creates a new file-based migration visitor.
-func newMeasureMigrationVisitor(group *commonv1.Group, shardNum, replicas
uint32, selector node.Selector, client queue.Client,
+func newMeasureMigrationVisitor(ctx context.Context, group *commonv1.Group,
shardNum, replicas uint32, selector node.Selector, client queue.Client,
l *logger.Logger, progress *Progress, chunkSize int,
targetStageInterval storage.IntervalRule, md metadata.Repo,
sourceStage, targetStage string, sourceSegmentInterval
storage.IntervalRule, orphanCfg orphanConfig,
) *measureMigrationVisitor {
return &measureMigrationVisitor{
+ ctx: ctx,
group: group.Metadata.Name,
sourceStage: sourceStage,
targetStage: targetStage,
@@ -284,10 +286,12 @@ func (mv *measureMigrationVisitor) VisitSeries(segmentTR
*timestamp.TimeRange, s
segmentIDStr := getSegmentTimeRange(targetSegmentTime,
mv.targetStageInterval).String()
for _, shardID := range shardIDs {
targetShardID :=
mv.calculateTargetShardID(uint32(shardID))
- partData :=
mv.createStreamingSegmentFromFiles(targetShardID, files, segmentTR,
data.TopicMeasureSeriesSync.String())
- // Stream segment to target shard replicas
- if err := mv.streamPartToTargetShard(partData); err !=
nil {
+ // Stream segment to target shard replicas. The factory
rebuilds the
+ // part on every retry so each attempt gets fresh
offset-0 readers.
+ if err := mv.streamPartToTargetShard(targetShardID,
func() queue.StreamingPartData {
+ return
mv.createStreamingSegmentFromFiles(targetShardID, files, segmentTR,
data.TopicMeasureSeriesSync.String())
+ }); err != nil {
errorMsg := fmt.Sprintf("failed to stream
measure segment to target shard %d: %v", targetShardID, err)
mv.recordError(scopeSeries, segmentTR, shardID,
nil, errorMsg)
return fmt.Errorf("failed to stream measure
segment to target shard %d: %w", targetShardID, err)
@@ -335,7 +339,7 @@ func (mv *measureMigrationVisitor) VisitPart(segmentTR
*timestamp.TimeRange, sou
// segment. Row-replay sidesteps this by publishing rows through the
// real write API, letting the receiver pick the target segment per row.
if len(targetSegments) > 1 {
- return mv.visitPartRowReplay(context.Background(), segmentTR,
sourceShardID, partID, partPath, targetSegments)
+ return mv.visitPartRowReplay(mv.ctx, segmentTR, sourceShardID,
partID, partPath, targetSegments)
}
atomic.AddUint64(&mv.partsCopiedSingleTarget, 1)
mv.progress.AddMeasureChunkSyncPart(mv.group)
@@ -370,19 +374,31 @@ func (mv *measureMigrationVisitor) VisitPart(segmentTR
*timestamp.TimeRange, sou
continue
}
- // Create file readers for this part
- files, release :=
measure.CreatePartFileReaderFromPath(partPath, mv.lfs)
- defer release()
-
- // Clone part data for this target segment
- targetPartData := partData
- targetPartData.Group = mv.group
- targetPartData.ShardID = targetShardID
- targetPartData.Topic = data.TopicMeasurePartSync.String()
- targetPartData.Files = files
+ // Reopen the part each attempt for fresh offset-0 readers,
releasing the
+ // prior attempt's handles first; the deferred call frees the
last set.
+ var prevRelease func()
+ defer func() {
+ if prevRelease != nil {
+ prevRelease()
+ }
+ }()
+ mk := func() queue.StreamingPartData {
+ if prevRelease != nil {
+ prevRelease()
+ prevRelease = nil
+ }
+ files, release :=
measure.CreatePartFileReaderFromPath(partPath, mv.lfs)
+ prevRelease = release
+ targetPartData := partData
+ targetPartData.Group = mv.group
+ targetPartData.ShardID = targetShardID
+ targetPartData.Topic =
data.TopicMeasurePartSync.String()
+ targetPartData.Files = files
+ return targetPartData
+ }
// Stream part to target segment
- if err := mv.streamPartToTargetShard(targetPartData); err !=
nil {
+ if err := mv.streamPartToTargetShard(targetShardID, mk); err !=
nil {
errorMsg := fmt.Sprintf("failed to stream measure part
to target segment %s: %v", targetSegmentTime.Format(time.RFC3339), err)
mv.recordError(scopePart, segmentTR, sourceShardID,
&partID, errorMsg)
return fmt.Errorf("failed to stream measure part to
target segment: %w", err)
@@ -491,22 +507,21 @@ func (mv *measureMigrationVisitor)
calculateTargetShardID(sourceShardID uint32)
return calculateTargetShardID(sourceShardID, mv.targetShardNum)
}
-// streamPartToTargetShard sends part data to all replicas of the target shard.
-func (mv *measureMigrationVisitor) streamPartToTargetShard(partData
queue.StreamingPartData) error {
- targetShardID := partData.ShardID
+// streamPartToTargetShard sends the part to every replica with bounded
+// exponential-backoff retry (transient: target restarting, disconnect,
+// receiver SERVER_BUSY). streamPartToNode closes the part's readers after each
+// send, so mk() is called per attempt to rebuild fresh offset-0 readers.
+func (mv *measureMigrationVisitor) streamPartToTargetShard(targetShardID
uint32, mk func() queue.StreamingPartData) error {
copies := mv.replicas + 1
// Send to all replicas using the exact pattern from steps.go:219-236
for replicaID := uint32(0); replicaID < copies; replicaID++ {
- // Use selector.Pick exactly like steps.go:220
- nodeID, err := mv.selector.Pick(mv.group, "", targetShardID,
replicaID)
+ err := pickAndRun(mv.ctx, mv.logger, mv.selector, mv.group, "",
targetShardID, replicaID, func(nodeID string) error {
+ partData := mk()
+ return mv.streamPartToNode(nodeID, partData.ShardID,
partData)
+ })
if err != nil {
- return fmt.Errorf("failed to pick node for shard %d
replica %d: %w", targetShardID, replicaID, err)
- }
-
- // Stream part data to target node using chunked sync
- if err := mv.streamPartToNode(nodeID, targetShardID, partData);
err != nil {
- return fmt.Errorf("failed to stream measure part to
node %s: %w", nodeID, err)
+ return fmt.Errorf("failed to stream measure part to
replica %d: %w", replicaID, err)
}
}
@@ -528,8 +543,7 @@ func (mv *measureMigrationVisitor) streamPartToNode(nodeID
string, targetShardID
}
// Stream using chunked transfer (same as syncer.go:202)
- ctx := context.Background()
- result, err := chunkedClient.SyncStreamingParts(ctx,
[]queue.StreamingPartData{partData})
+ result, err := chunkedClient.SyncStreamingParts(mv.ctx,
[]queue.StreamingPartData{partData})
if err != nil {
return fmt.Errorf("failed to sync streaming parts to node %s:
%w", nodeID, err)
}
diff --git a/banyand/backup/lifecycle/migration_error_test.go
b/banyand/backup/lifecycle/migration_error_test.go
index ead26b7c9..05aa8782c 100644
--- a/banyand/backup/lifecycle/migration_error_test.go
+++ b/banyand/backup/lifecycle/migration_error_test.go
@@ -18,6 +18,7 @@
package lifecycle
import (
+ "context"
"encoding/json"
"fmt"
"sync"
@@ -98,7 +99,7 @@ func TestVisitorRecordError(t *testing.T) {
t.Run("measure_part_with_shard_zero", func(t *testing.T) {
p := NewProgress("", logger.GetLogger("test"))
- mv := newMeasureMigrationVisitor(grp, 1, 0, nil, nil,
logger.GetLogger("test"), p, 0, dayInterval, nil,
+ mv := newMeasureMigrationVisitor(context.Background(), grp, 1,
0, nil, nil, logger.GetLogger("test"), p, 0, dayInterval, nil,
"hot", "warm", dayInterval, orphanConfig{})
part := uint64(1)
mv.recordError(scopePart, segTR, 0, &part, "boom")
@@ -120,7 +121,7 @@ func TestVisitorRecordError(t *testing.T) {
t.Run("measure_series_has_no_part", func(t *testing.T) {
p := NewProgress("", logger.GetLogger("test"))
- mv := newMeasureMigrationVisitor(grp, 1, 0, nil, nil,
logger.GetLogger("test"), p, 0, dayInterval, nil,
+ mv := newMeasureMigrationVisitor(context.Background(), grp, 1,
0, nil, nil, logger.GetLogger("test"), p, 0, dayInterval, nil,
"hot", "warm", dayInterval, orphanConfig{})
mv.recordError(scopeSeries, segTR, 3, nil, "series boom")
require.Len(t, p.MigrationErrors, 1)
@@ -133,7 +134,7 @@ func TestVisitorRecordError(t *testing.T) {
t.Run("trace_shard_catalog", func(t *testing.T) {
p := NewProgress("", logger.GetLogger("test"))
- tv := newTraceMigrationVisitor(grp, 1, 0, nil, nil,
logger.GetLogger("test"), p, 0, dayInterval, nil,
+ tv := newTraceMigrationVisitor(context.Background(), grp, 1, 0,
nil, nil, logger.GetLogger("test"), p, 0, dayInterval, nil,
"warm", "cold", dayInterval)
tv.recordError(scopeShard, segTR, 2, "trace boom")
require.Len(t, p.MigrationErrors, 1)
diff --git a/banyand/backup/lifecycle/migration_integration.go
b/banyand/backup/lifecycle/migration_integration.go
index 25efcd57b..f4f278bc8 100644
--- a/banyand/backup/lifecycle/migration_integration.go
+++ b/banyand/backup/lifecycle/migration_integration.go
@@ -18,6 +18,7 @@
package lifecycle
import (
+ "context"
"strings"
"github.com/apache/skywalking-banyandb/api/common"
@@ -32,7 +33,7 @@ import (
)
// migrateStreamWithFileBasedAndProgress performs file-based stream migration
with progress tracking.
-func migrateStreamWithFileBasedAndProgress(tsdbRootPath string, timeRange
timestamp.TimeRange, group *GroupConfig,
+func migrateStreamWithFileBasedAndProgress(ctx context.Context, tsdbRootPath
string, timeRange timestamp.TimeRange, group *GroupConfig,
logger *logger.Logger, progress *Progress, chunkSize int, md
metadata.Repo, orphanCfg orphanConfig,
) ([]string, error) {
// Convert segment Interval to IntervalRule using
storage.MustToIntervalRule
@@ -55,7 +56,7 @@ func migrateStreamWithFileBasedAndProgress(tsdbRootPath
string, timeRange timest
// Create file-based migration visitor with progress tracking and
target stage interval
visitor := newStreamMigrationVisitor(
- group.Group, group.TargetShardNum, group.TargetReplicas,
group.NodeSelector, group.QueueClient,
+ ctx, group.Group, group.TargetShardNum, group.TargetReplicas,
group.NodeSelector, group.QueueClient,
logger, progress, chunkSize, targetStageInterval, md,
group.SourceStage, group.TargetStage, segmentIntervalRule,
orphanCfg,
)
@@ -125,7 +126,7 @@ func (pcv *partCountVisitor) VisitElementIndex(_
*timestamp.TimeRange, _ common.
}
// migrateMeasureWithFileBasedAndProgress performs file-based measure
migration with progress tracking.
-func migrateMeasureWithFileBasedAndProgress(tsdbRootPath string, timeRange
timestamp.TimeRange, group *GroupConfig,
+func migrateMeasureWithFileBasedAndProgress(ctx context.Context, tsdbRootPath
string, timeRange timestamp.TimeRange, group *GroupConfig,
logger *logger.Logger, progress *Progress, chunkSize int, md
metadata.Repo, orphanCfg orphanConfig,
) ([]string, error) {
// Convert segment interval to IntervalRule using
storage.MustToIntervalRule
@@ -147,7 +148,7 @@ func migrateMeasureWithFileBasedAndProgress(tsdbRootPath
string, timeRange times
// Create file-based migration visitor with progress tracking and
target stage interval
visitor := newMeasureMigrationVisitor(
- group.Group, group.TargetShardNum, group.TargetReplicas,
group.NodeSelector, group.QueueClient,
+ ctx, group.Group, group.TargetShardNum, group.TargetReplicas,
group.NodeSelector, group.QueueClient,
logger, progress, chunkSize, targetStageInterval, md,
group.SourceStage, group.TargetStage, segmentIntervalRule,
orphanCfg,
)
@@ -245,7 +246,7 @@ func (pcv *measurePartCountVisitor) VisitPart(_
*timestamp.TimeRange, _ common.S
}
// migrateTraceWithFileBasedAndProgress performs file-based trace migration
with progress tracking.
-func migrateTraceWithFileBasedAndProgress(tsdbRootPath string, timeRange
timestamp.TimeRange, group *GroupConfig,
+func migrateTraceWithFileBasedAndProgress(ctx context.Context, tsdbRootPath
string, timeRange timestamp.TimeRange, group *GroupConfig,
logger *logger.Logger, progress *Progress, chunkSize int, md
metadata.Repo,
) ([]string, error) {
// Convert segment interval to IntervalRule using
storage.MustToIntervalRule
@@ -267,7 +268,7 @@ func migrateTraceWithFileBasedAndProgress(tsdbRootPath
string, timeRange timesta
// Create file-based migration visitor with progress tracking and
target stage interval
visitor := newTraceMigrationVisitor(
- group.Group, group.TargetShardNum, group.TargetReplicas,
group.NodeSelector, group.QueueClient,
+ ctx, group.Group, group.TargetShardNum, group.TargetReplicas,
group.NodeSelector, group.QueueClient,
logger, progress, chunkSize, targetStageInterval, md,
group.SourceStage, group.TargetStage, segmentIntervalRule,
)
diff --git a/banyand/backup/lifecycle/retry.go
b/banyand/backup/lifecycle/retry.go
new file mode 100644
index 000000000..3f7a6a473
--- /dev/null
+++ b/banyand/backup/lifecycle/retry.go
@@ -0,0 +1,135 @@
+// 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 lifecycle
+
+import (
+ "context"
+ "errors"
+ "fmt"
+ "strings"
+ "time"
+
+ "github.com/cenkalti/backoff/v4"
+
+ "github.com/apache/skywalking-banyandb/banyand/queue"
+ "github.com/apache/skywalking-banyandb/pkg/grpchelper"
+ "github.com/apache/skywalking-banyandb/pkg/logger"
+ "github.com/apache/skywalking-banyandb/pkg/node"
+)
+
+// defaultLifecycleSendRetryTimeout bounds the total wall-clock spent retrying
+// transient failures when streaming a part to a target node (e.g. while the
+// target node restarts). Configurable via --lifecycle-send-retry-timeout.
+const defaultLifecycleSendRetryTimeout = 15 * time.Minute
+
+// lifecycleSendRetryTimeout is the active bound, set by the lifecycle flag.
+var lifecycleSendRetryTimeout = defaultLifecycleSendRetryTimeout
+
+// isNoNodes reports whether err means the target node could not be picked
+// because no node is currently available (e.g. the target is restarting).
+// It matches both the typed sentinel and the plain round-robin error string.
+func isNoNodes(err error) bool {
+ if errors.Is(err, node.ErrNoAvailableNode) {
+ return true
+ }
+ return err != nil && strings.Contains(err.Error(), "no nodes available")
+}
+
+// isTransientSend reports whether err from streaming a part to a node is
+// transient and worth retrying: receiver back-pressure (SERVER_BUSY) or a
+// transient/failover gRPC error such as Unavailable, ResourceExhausted, or a
+// dropped connection.
+func isTransientSend(err error) bool {
+ return errors.Is(err, queue.ErrServerBusy) ||
grpchelper.IsTransientError(err) || grpchelper.IsFailoverError(err)
+}
+
+// retryLogInterval throttles the "still retrying" heartbeat so a long wait for
+// a restarting node logs at most once per minute instead of on every attempt.
+const retryLogInterval = time.Minute
+
+// nodePicker is the one method of node.Selector that the send-retry path
needs;
+// narrowing it keeps runWithRetry/pickWithRetry decoupled and easy to test.
+type nodePicker interface {
+ Pick(group, name string, shardID, replicaID uint32) (string, error)
+}
+
+// runWithRetry runs op under bounded exponential backoff capped by
+// lifecycleSendRetryTimeout. op returns a transient error to retry or
+// backoff.Permanent(err) to stop immediately. While retrying it logs a
+// heartbeat at most once per minute so a long wait for a restarting node is
not
+// mistaken for a hang. ctx cancellation aborts the retry promptly (returning
+// ctx.Err()) instead of waiting out the timeout.
+func runWithRetry(ctx context.Context, l *logger.Logger, stage string, op
func() error) error {
+ bo := backoff.NewExponentialBackOff()
+ bo.MaxElapsedTime = lifecycleSendRetryTimeout
+ start := time.Now()
+ var lastLog time.Time
+ return backoff.RetryNotify(op, backoff.WithContext(bo, ctx), func(err
error, _ time.Duration) {
+ now := time.Now()
+ if !lastLog.IsZero() && now.Sub(lastLog) < retryLogInterval {
+ return
+ }
+ lastLog = now
+ l.Warn().Err(err).Str("stage", stage).Dur("elapsed",
now.Sub(start)).
+ Msg("transient lifecycle send failure, still retrying")
+ })
+}
+
+// pickWithRetry picks a target node, retrying on "no nodes available" (target
+// restarting) under runWithRetry's bounded backoff and heartbeat logging.
+func pickWithRetry(ctx context.Context, l *logger.Logger, p nodePicker, group,
name string, shardID uint32) (string, error) {
+ var nodeID string
+ err := runWithRetry(ctx, l, fmt.Sprintf("pick group=%s name=%s
shard=%d", group, name, shardID), func() error {
+ var pErr error
+ nodeID, pErr = p.Pick(group, name, shardID, 0)
+ switch {
+ case pErr == nil:
+ return nil
+ case isNoNodes(pErr):
+ return pErr
+ default:
+ return backoff.Permanent(pErr)
+ }
+ })
+ return nodeID, err
+}
+
+// pickAndRun picks a target node and runs run against it inside one bounded
+// backoff loop: a "no nodes" pick error or a transient run error retries the
+// whole pick+run (so a failed send re-picks, possibly a healthier node); any
+// other error stops immediately. Heartbeat-logged like runWithRetry.
+func pickAndRun(ctx context.Context, l *logger.Logger, p nodePicker, group,
name string, shardID, replicaID uint32,
+ run func(nodeID string) error,
+) error {
+ return runWithRetry(ctx, l, fmt.Sprintf("send group=%s name=%s shard=%d
replica=%d", group, name, shardID, replicaID), func() error {
+ nodeID, pickErr := p.Pick(group, name, shardID, replicaID)
+ if pickErr != nil {
+ if isNoNodes(pickErr) {
+ return pickErr
+ }
+ return backoff.Permanent(pickErr)
+ }
+ if runErr := run(nodeID); runErr != nil {
+ if isTransientSend(runErr) {
+ return runErr
+ }
+ return backoff.Permanent(runErr)
+ }
+ return nil
+ })
+}
diff --git a/banyand/backup/lifecycle/retry_test.go
b/banyand/backup/lifecycle/retry_test.go
new file mode 100644
index 000000000..7d03c78dd
--- /dev/null
+++ b/banyand/backup/lifecycle/retry_test.go
@@ -0,0 +1,132 @@
+// 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 lifecycle
+
+import (
+ "context"
+ "errors"
+ "fmt"
+ "testing"
+
+ "github.com/stretchr/testify/assert"
+
+ "github.com/apache/skywalking-banyandb/banyand/queue"
+ "github.com/apache/skywalking-banyandb/pkg/logger"
+ "github.com/apache/skywalking-banyandb/pkg/node"
+)
+
+func TestIsNoNodes(t *testing.T) {
+ tests := []struct {
+ err error
+ name string
+ want bool
+ }{
+ {name: "typed sentinel", err: node.ErrNoAvailableNode, want:
true},
+ {name: "wrapped sentinel", err: fmt.Errorf("pick failed: %w",
node.ErrNoAvailableNode), want: true},
+ {name: "plain round-robin string", err: errors.New("no nodes
available"), want: true},
+ {name: "wrapped round-robin string", err: fmt.Errorf("replica
0: %w", errors.New("no nodes available")), want: true},
+ {name: "unrelated error", err: errors.New("connection
refused"), want: false},
+ {name: "nil error", err: nil, want: false},
+ }
+ for _, tt := range tests {
+ t.Run(tt.name, func(t *testing.T) {
+ assert.Equal(t, tt.want, isNoNodes(tt.err))
+ })
+ }
+}
+
+func TestIsTransientSend(t *testing.T) {
+ tests := []struct {
+ err error
+ name string
+ want bool
+ }{
+ {name: "server busy sentinel", err: queue.ErrServerBusy, want:
true},
+ {name: "wrapped server busy", err: fmt.Errorf("receiver busy on
node x: %w", queue.ErrServerBusy), want: true},
+ {name: "plain non-transient", err: errors.New("permission
denied"), want: false},
+ {name: "nil error", err: nil, want: false},
+ }
+ for _, tt := range tests {
+ t.Run(tt.name, func(t *testing.T) {
+ assert.Equal(t, tt.want, isTransientSend(tt.err))
+ })
+ }
+}
+
+// fakePicker implements nodePicker: it returns "no nodes available" the first
+// noNodes times, then failErr (if set), then node.
+type fakePicker struct {
+ failErr error
+ node string
+ noNodes int
+ calls int
+}
+
+func (f *fakePicker) Pick(_, _ string, _, _ uint32) (string, error) {
+ f.calls++
+ if f.noNodes > 0 {
+ f.noNodes--
+ return "", errors.New("no nodes available")
+ }
+ if f.failErr != nil {
+ return "", f.failErr
+ }
+ return f.node, nil
+}
+
+func TestPickWithRetry(t *testing.T) {
+ l := logger.GetLogger("test")
+
+ t.Run("success first try", func(t *testing.T) {
+ p := &fakePicker{node: "n1"}
+ got, err := pickWithRetry(context.Background(), l, p, "g", "n",
0)
+ assert.NoError(t, err)
+ assert.Equal(t, "n1", got)
+ assert.Equal(t, 1, p.calls)
+ })
+
+ t.Run("permanent error fails fast without retry", func(t *testing.T) {
+ boom := errors.New("unknown shard")
+ p := &fakePicker{failErr: boom}
+ got, err := pickWithRetry(context.Background(), l, p, "g", "n",
0)
+ assert.ErrorIs(t, err, boom)
+ assert.Empty(t, got)
+ assert.Equal(t, 1, p.calls)
+ })
+
+ t.Run("no nodes then recover retries", func(t *testing.T) {
+ p := &fakePicker{node: "n2", noNodes: 1}
+ got, err := pickWithRetry(context.Background(), l, p, "g", "n",
0)
+ assert.NoError(t, err)
+ assert.Equal(t, "n2", got)
+ assert.Equal(t, 2, p.calls)
+ })
+
+ t.Run("canceled context aborts retry promptly", func(t *testing.T) {
+ ctx, cancel := context.WithCancel(context.Background())
+ cancel()
+ // noNodes is far higher than any attempt count: without
honoring ctx the
+ // call would keep retrying until lifecycleSendRetryTimeout.
With ctx
+ // canceled it must return context.Canceled after a single
attempt.
+ p := &fakePicker{node: "n3", noNodes: 1000}
+ got, err := pickWithRetry(ctx, l, p, "g", "n", 0)
+ assert.ErrorIs(t, err, context.Canceled)
+ assert.Empty(t, got)
+ assert.Equal(t, 1, p.calls)
+ })
+}
diff --git a/banyand/backup/lifecycle/row_replay_measure.go
b/banyand/backup/lifecycle/row_replay_measure.go
index 6f34a8002..64270bc62 100644
--- a/banyand/backup/lifecycle/row_replay_measure.go
+++ b/banyand/backup/lifecycle/row_replay_measure.go
@@ -67,6 +67,7 @@ type cachedMeasureSchema struct {
// measureRowReplayer replays a group's measure parts row-by-row through the
// real Write API, resolving each row's schema on demand via EntityValues.
type measureRowReplayer struct {
+ ctx context.Context
selector node.Selector
archiver *orphanArchiver
sender *batchSender
@@ -118,6 +119,7 @@ func newMeasureRowReplayer(
measureSchemas[m.GetMetadata().GetName()] = m
}
return &measureRowReplayer{
+ ctx: ctx,
group: group,
targetShardNum: targetShardNum,
selector: selector,
@@ -347,7 +349,7 @@ func (r *measureRowReplayer) routeColumnar(
if rErr != nil {
return 0, nil, "", fmt.Errorf("route %s: %w",
bc.subject, rErr)
}
- pickedNode, pErr := r.selector.Pick(r.group,
bc.subject, uint32(sid), 0)
+ pickedNode, pErr := pickWithRetry(r.ctx, r.logger,
r.selector, r.group, bc.subject, uint32(sid))
if pErr != nil {
return 0, nil, "", fmt.Errorf("pick target node
for %s: %w", bc.subject, pErr)
}
@@ -363,7 +365,7 @@ func (r *measureRowReplayer) routeColumnar(
if rErr != nil {
return 0, nil, "", fmt.Errorf("route %s: %w", bc.subject, rErr)
}
- nodeID, pErr := r.selector.Pick(r.group, bc.subject, uint32(sid), 0)
+ nodeID, pErr := pickWithRetry(r.ctx, r.logger, r.selector, r.group,
bc.subject, uint32(sid))
if pErr != nil {
return 0, nil, "", fmt.Errorf("pick target node for %s: %w",
bc.subject, pErr)
}
diff --git a/banyand/backup/lifecycle/row_replay_pipeline.go
b/banyand/backup/lifecycle/row_replay_pipeline.go
index 7b3fc3d94..7ad9afd4d 100644
--- a/banyand/backup/lifecycle/row_replay_pipeline.go
+++ b/banyand/backup/lifecycle/row_replay_pipeline.go
@@ -302,10 +302,10 @@ func (s *batchSender) enqueueMarshaled(ctx
context.Context, nodeID, group string
// routeAndEnqueue picks the target node for iwr and enqueues it. The
// proto-message replayers (stream/trace) share this Pick+build+enqueue tail;
the
// measure columnar path routes inline via routeColumnar + enqueueMarshaled.
-func (s *batchSender) routeAndEnqueue(ctx context.Context, selector
node.Selector,
+func (s *batchSender) routeAndEnqueue(ctx context.Context, l *logger.Logger,
selector node.Selector,
group, name string, shardID uint32, iwr proto.Message,
) error {
- nodeID, err := selector.Pick(group, name, shardID, 0)
+ nodeID, err := pickWithRetry(ctx, l, selector, group, name, shardID)
if err != nil {
return fmt.Errorf("pick target node for %s: %w", name, err)
}
diff --git a/banyand/backup/lifecycle/row_replay_stream.go
b/banyand/backup/lifecycle/row_replay_stream.go
index 471a18ba1..408e93c95 100644
--- a/banyand/backup/lifecycle/row_replay_stream.go
+++ b/banyand/backup/lifecycle/row_replay_stream.go
@@ -234,7 +234,7 @@ func (r *streamRowReplayer) publishRow(ctx context.Context,
pw *orphanPartWriter
if err != nil {
return err
}
- return r.sender.routeAndEnqueue(ctx, r.selector, r.group,
wr.Metadata.Name, iwr.ShardId, iwr)
+ return r.sender.routeAndEnqueue(ctx, r.logger, r.selector, r.group,
wr.Metadata.Name, iwr.ShardId, iwr)
}
// archiveOrphanRow writes one orphan stream row to the part archive (a no-op
diff --git a/banyand/backup/lifecycle/row_replay_trace.go
b/banyand/backup/lifecycle/row_replay_trace.go
index fd2d4dd52..919ffcc20 100644
--- a/banyand/backup/lifecycle/row_replay_trace.go
+++ b/banyand/backup/lifecycle/row_replay_trace.go
@@ -136,5 +136,5 @@ func (r *traceRowReplayer) buildWriteRequest(row
dumptrace.Row) (*tracev1.WriteR
func (r *traceRowReplayer) publishRow(ctx context.Context, row dumptrace.Row)
error {
_, iwr := r.buildWriteRequest(row)
- return r.sender.routeAndEnqueue(ctx, r.selector, r.group, r.traceName,
iwr.ShardId, iwr)
+ return r.sender.routeAndEnqueue(ctx, r.logger, r.selector, r.group,
r.traceName, iwr.ShardId, iwr)
}
diff --git a/banyand/backup/lifecycle/service.go
b/banyand/backup/lifecycle/service.go
index 693723464..ed2433515 100644
--- a/banyand/backup/lifecycle/service.go
+++ b/banyand/backup/lifecycle/service.go
@@ -102,6 +102,7 @@ type lifecycleService struct {
sch *timestamp.Scheduler
stopCh chan struct{}
clientCloser context.CancelFunc
+ runCancel context.CancelFunc
currentNode *databasev1.Node
tlsReloader *pkgtls.Reloader
httpSrv *http.Server
@@ -212,6 +213,8 @@ func (l *lifecycleService) FlagSet() *run.FlagSet {
"Maximum rows per row-replay batch (row-replay is the fallback
for parts spanning multiple target segments)")
flagS.VarP(&rowReplayMaxBatchBytes, "row-replay-max-batch-bytes", "",
"Maximum in-flight marshaled bytes per row-replay batch; caps
peak marshal memory for large bodies (default: 32MB)")
+ flagS.DurationVar(&lifecycleSendRetryTimeout,
"lifecycle-send-retry-timeout", defaultLifecycleSendRetryTimeout,
+ "Max total time to retry streaming a migration part to a target
node on transient failures (target restarting, disconnect, receiver busy)")
// Lifecycle server flags
flagS.BoolVar(&l.lifecycleTLS, "lifecycle-tls", false, "connection uses
TLS if true, else plain TCP")
@@ -317,6 +320,10 @@ func (l *lifecycleService) GracefulStop() {
l.tlsReloader.Stop()
}
+ if l.runCancel != nil {
+ l.runCancel()
+ }
+
if l.clientCloser != nil {
l.clientCloser()
}
@@ -438,7 +445,12 @@ func (l *lifecycleService) Serve() run.StopNotify {
if l.schedule == "" {
defer close(done)
l.l.Info().Msg("starting lifecycle migration without schedule")
- if err := l.action(context.Background()); err != nil {
+ // One-shot mode has no scheduler, so derive a cancellable
context here
+ // that GracefulStop can cancel to abort an in-flight migration
promptly.
+ ctx, cancel := context.WithCancel(context.Background())
+ l.runCancel = cancel
+ defer cancel()
+ if err := l.action(ctx); err != nil {
logger.Panicf("failed to run lifecycle migration: %v",
err)
}
return done
@@ -449,6 +461,8 @@ func (l *lifecycleService) Serve() run.StopNotify {
var executionCount int
err := l.sch.Register(context.Background(), "lifecycle",
cron.Descriptor, l.schedule, func(ctx context.Context, triggerTime time.Time, _
*logger.Logger) bool {
l.l.Info().Msgf("lifecycle migration triggered at %s",
triggerTime)
+ // The scheduler cancels this context on Close (GracefulStop),
which lets
+ // the send path abort promptly; its 5-minute wait does not
cancel it.
if err := l.action(ctx); err != nil {
l.l.Error().Err(err).Msg("failed to run lifecycle
migration action")
}
@@ -1215,7 +1229,7 @@ func (l *lifecycleService) processStreamGroup(ctx
context.Context, g *commonv1.G
}
// processStreamGroupFileBased uses file-based migration instead of
element-based queries.
-func (l *lifecycleService) processStreamGroupFileBased(_ context.Context, g
*GroupConfig, streamDir string,
+func (l *lifecycleService) processStreamGroupFileBased(ctx context.Context, g
*GroupConfig, streamDir string,
tr *timestamp.TimeRange, progress *Progress,
) ([]string, error) {
if progress.IsStreamGroupDeleted(g.Metadata.Name) {
@@ -1234,8 +1248,7 @@ func (l *lifecycleService) processStreamGroupFileBased(_
context.Context, g *Gro
}
// Use the file-based migration with existing visitor pattern
- //nolint:contextcheck // migration drives its own context lifecycle for
batch publish.
- segmentSuffixes, err := migrateStreamWithFileBasedAndProgress(rootDir,
*tr, g, l.l, progress, int(l.chunkSize), l.metadata,
l.orphanConfigFor(l.streamRoot))
+ segmentSuffixes, err := migrateStreamWithFileBasedAndProgress(ctx,
rootDir, *tr, g, l.l, progress, int(l.chunkSize), l.metadata,
l.orphanConfigFor(l.streamRoot))
if err != nil {
return nil, fmt.Errorf("file-based stream migration failed:
%w", err)
}
@@ -1340,7 +1353,7 @@ func (l *lifecycleService) processMeasureGroup(ctx
context.Context, g *commonv1.
}
// processMeasureGroupFileBased uses file-based migration instead of
query-based migration.
-func (l *lifecycleService) processMeasureGroupFileBased(_ context.Context, g
*GroupConfig, measureDir string,
+func (l *lifecycleService) processMeasureGroupFileBased(ctx context.Context, g
*GroupConfig, measureDir string,
tr *timestamp.TimeRange, progress *Progress,
) ([]string, error) {
if progress.IsMeasureGroupDeleted(g.Metadata.Name) {
@@ -1359,8 +1372,7 @@ func (l *lifecycleService) processMeasureGroupFileBased(_
context.Context, g *Gr
}
// Use the file-based migration with existing visitor pattern
- //nolint:contextcheck // migration drives its own context lifecycle for
batch publish.
- segmentSuffixes, err := migrateMeasureWithFileBasedAndProgress(rootDir,
*tr, g, l.l, progress, int(l.chunkSize), l.metadata,
l.orphanConfigFor(l.measureRoot))
+ segmentSuffixes, err := migrateMeasureWithFileBasedAndProgress(ctx,
rootDir, *tr, g, l.l, progress, int(l.chunkSize), l.metadata,
l.orphanConfigFor(l.measureRoot))
if err != nil {
return nil, fmt.Errorf("file-based measure migration failed:
%w", err)
}
@@ -1447,7 +1459,7 @@ func (l *lifecycleService) processTraceGroup(ctx
context.Context, g *commonv1.Gr
progress.Save(l.progressFilePath, l.l)
}
-func (l *lifecycleService) processTraceGroupFileBased(_ context.Context, g
*GroupConfig, traceDir string,
+func (l *lifecycleService) processTraceGroupFileBased(ctx context.Context, g
*GroupConfig, traceDir string,
tr *timestamp.TimeRange, progress *Progress,
) ([]string, error) {
if progress.IsTraceGroupDeleted(g.Metadata.Name) {
@@ -1466,8 +1478,7 @@ func (l *lifecycleService) processTraceGroupFileBased(_
context.Context, g *Grou
}
// Use the file-based migration with existing visitor pattern
- //nolint:contextcheck // migration drives its own context lifecycle for
batch publish.
- segmentSuffixes, err := migrateTraceWithFileBasedAndProgress(rootDir,
*tr, g, l.l, progress, int(l.chunkSize), l.metadata)
+ segmentSuffixes, err := migrateTraceWithFileBasedAndProgress(ctx,
rootDir, *tr, g, l.l, progress, int(l.chunkSize), l.metadata)
if err != nil {
return nil, fmt.Errorf("file-based trace migration failed: %w",
err)
}
diff --git a/banyand/backup/lifecycle/stream_migration_visitor.go
b/banyand/backup/lifecycle/stream_migration_visitor.go
index 655583de6..d22ec583a 100644
--- a/banyand/backup/lifecycle/stream_migration_visitor.go
+++ b/banyand/backup/lifecycle/stream_migration_visitor.go
@@ -41,6 +41,7 @@ import (
// streamMigrationVisitor implements the stream.Visitor interface for
file-based migration.
type streamMigrationVisitor struct {
+ ctx context.Context
client queue.Client
lfs fs.FileSystem
metadata metadata.Repo
@@ -64,11 +65,12 @@ type streamMigrationVisitor struct {
}
// newStreamMigrationVisitor creates a new file-based migration visitor.
-func newStreamMigrationVisitor(group *commonv1.Group, shardNum, replicas
uint32, selector node.Selector, client queue.Client,
+func newStreamMigrationVisitor(ctx context.Context, group *commonv1.Group,
shardNum, replicas uint32, selector node.Selector, client queue.Client,
l *logger.Logger, progress *Progress, chunkSize int,
targetStageInterval storage.IntervalRule, md metadata.Repo,
sourceStage, targetStage string, sourceSegmentInterval
storage.IntervalRule, orphanCfg orphanConfig,
) *streamMigrationVisitor {
return &streamMigrationVisitor{
+ ctx: ctx,
group: group.Metadata.Name,
sourceStage: sourceStage,
targetStage: targetStage,
@@ -260,17 +262,11 @@ func (mv *streamMigrationVisitor) VisitSeries(segmentTR
*timestamp.TimeRange, se
segmentIDStr := getSegmentTimeRange(targetSegmentTime,
mv.targetStageInterval).String()
for _, shardID := range shardIDs {
targetShardID :=
mv.calculateTargetShardID(uint32(shardID))
- ff := make([]queue.FileInfo, 0, len(files))
- for _, file := range files {
- ff = append(ff, queue.FileInfo{
- Name: file.name,
- Reader: file.file.SequentialRead(),
- })
- }
- partData :=
mv.createStreamingSegmentFromFiles(targetShardID, ff, segmentTR,
data.TopicStreamSeriesSync.String())
-
- // Stream segment to target shard replicas
- if err := mv.streamPartToTargetShard(partData); err !=
nil {
+ // Stream segment to target shard replicas. The factory
rebuilds the
+ // part on every retry so each attempt gets fresh
offset-0 readers.
+ if err := mv.streamPartToTargetShard(targetShardID,
func() queue.StreamingPartData {
+ return
mv.createStreamingSegmentFromFiles(targetShardID, files, segmentTR,
data.TopicStreamSeriesSync.String())
+ }); err != nil {
errorMsg := fmt.Sprintf("failed to stream
segment to target shard %d: %v", targetShardID, err)
mv.recordError(scopeSeries, segmentTR, shardID,
nil, errorMsg)
return fmt.Errorf("failed to stream segment to
target shard %d: %w", targetShardID, err)
@@ -321,7 +317,7 @@ func (mv *streamMigrationVisitor) VisitPart(segmentTR
*timestamp.TimeRange, sour
Msg("migrating stream part")
if len(targetSegments) > 1 {
- return mv.visitPartRowReplay(context.Background(), segmentTR,
sourceShardID, partID, partPath, targetSegments)
+ return mv.visitPartRowReplay(mv.ctx, segmentTR, sourceShardID,
partID, partPath, targetSegments)
}
atomic.AddUint64(&mv.partsCopiedSingleTarget, 1)
mv.progress.AddStreamChunkSyncPart(mv.group)
@@ -347,19 +343,31 @@ func (mv *streamMigrationVisitor) VisitPart(segmentTR
*timestamp.TimeRange, sour
continue
}
- // Create file readers for this part
- files, release := stream.CreatePartFileReaderFromPath(partPath,
mv.lfs)
- defer release()
-
- // Clone part data for this target segment
- targetPartData := partData
- targetPartData.Group = mv.group
- targetPartData.ShardID = targetShardID
- targetPartData.Topic = data.TopicStreamPartSync.String()
- targetPartData.Files = files
+ // Reopen the part each attempt for fresh offset-0 readers,
releasing the
+ // prior attempt's handles first; the deferred call frees the
last set.
+ var prevRelease func()
+ defer func() {
+ if prevRelease != nil {
+ prevRelease()
+ }
+ }()
+ mk := func() queue.StreamingPartData {
+ if prevRelease != nil {
+ prevRelease()
+ prevRelease = nil
+ }
+ files, release :=
stream.CreatePartFileReaderFromPath(partPath, mv.lfs)
+ prevRelease = release
+ targetPartData := partData
+ targetPartData.Group = mv.group
+ targetPartData.ShardID = targetShardID
+ targetPartData.Topic = data.TopicStreamPartSync.String()
+ targetPartData.Files = files
+ return targetPartData
+ }
// Stream part to target segment
- if err := mv.streamPartToTargetShard(targetPartData); err !=
nil {
+ if err := mv.streamPartToTargetShard(targetShardID, mk); err !=
nil {
errorMsg := fmt.Sprintf("failed to stream part to
target segment %s: %v", targetSegmentTime.Format(time.RFC3339), err)
mv.recordError(scopePart, segmentTR, sourceShardID,
&partID, errorMsg)
return fmt.Errorf("failed to stream part to target
segment: %w", err)
@@ -520,8 +528,9 @@ func (mv *streamMigrationVisitor)
VisitElementIndex(segmentTR *timestamp.TimeRan
Uint32("target_shard", targetShardID).
Msg("found element index segment files for migration")
- // Create FileInfo for this segment file
- files := make([]queue.FileInfo, 0, len(segmentFiles))
+ // Keep the open segment file handles so the send factory can rebuild
+ // fresh offset-0 readers on every retry via SequentialRead.
+ openFiles := make([]fileInfo, 0, len(segmentFiles))
// Process each segment file
for _, segmentFileName := range segmentFiles {
// Extract segment ID from filename (remove .seg extension)
@@ -558,9 +567,9 @@ func (mv *streamMigrationVisitor)
VisitElementIndex(segmentTR *timestamp.TimeRan
// Close the file reader
defer segmentFile.Close()
- files = append(files, queue.FileInfo{
- Name: segmentFileName,
- Reader: segmentFile.SequentialRead(),
+ openFiles = append(openFiles, fileInfo{
+ file: segmentFile,
+ name: segmentFileName,
})
mv.logger.Info().
@@ -571,10 +580,12 @@ func (mv *streamMigrationVisitor)
VisitElementIndex(segmentTR *timestamp.TimeRan
Int("total_segments",
mv.progress.GetStreamElementIndexCount(mv.group)).
Msg("element index segment migration completed
successfully")
}
- partData := mv.createStreamingSegmentFromFiles(targetShardID, files,
segmentTR, data.TopicStreamElementIndexSync.String())
- // Stream segment file to target shard replicas
- if err := mv.streamPartToTargetShard(partData); err != nil {
+ // Stream segment file to target shard replicas. The factory rebuilds
the
+ // part on every retry so each attempt gets fresh offset-0 readers.
+ if err := mv.streamPartToTargetShard(targetShardID, func()
queue.StreamingPartData {
+ return mv.createStreamingSegmentFromFiles(targetShardID,
openFiles, segmentTR, data.TopicStreamElementIndexSync.String())
+ }); err != nil {
errorMsg := fmt.Sprintf("failed to stream element index to
target shard: %v", err)
mv.recordError(scopeElementIndex, segmentTR, sourceShardID,
nil, errorMsg)
return fmt.Errorf("failed to stream element index to target
shard: %w", err)
@@ -592,22 +603,21 @@ func (mv *streamMigrationVisitor)
calculateTargetShardID(sourceShardID uint32) u
return calculateTargetShardID(sourceShardID, mv.targetShardNum)
}
-// streamPartToTargetShard sends part data to all replicas of the target shard.
-func (mv *streamMigrationVisitor) streamPartToTargetShard(partData
queue.StreamingPartData) error {
- targetShardID := partData.ShardID
+// streamPartToTargetShard sends the part to every replica with bounded
+// exponential-backoff retry (transient: target restarting, disconnect,
+// receiver SERVER_BUSY). streamPartToNode closes the part's readers after each
+// send, so mk() is called per attempt to rebuild fresh offset-0 readers.
+func (mv *streamMigrationVisitor) streamPartToTargetShard(targetShardID
uint32, mk func() queue.StreamingPartData) error {
copies := mv.replicas + 1
// Send to all replicas using the exact pattern from steps.go:219-236
for replicaID := uint32(0); replicaID < copies; replicaID++ {
- // Use selector.Pick exactly like steps.go:220
- nodeID, err := mv.selector.Pick(mv.group, "", targetShardID,
replicaID)
+ err := pickAndRun(mv.ctx, mv.logger, mv.selector, mv.group, "",
targetShardID, replicaID, func(nodeID string) error {
+ partData := mk()
+ return mv.streamPartToNode(nodeID, partData.ShardID,
partData)
+ })
if err != nil {
- return fmt.Errorf("failed to pick node for shard %d
replica %d: %w", targetShardID, replicaID, err)
- }
-
- // Stream part data to target node using chunked sync
- if err := mv.streamPartToNode(nodeID, targetShardID, partData);
err != nil {
- return fmt.Errorf("failed to stream part to node %s:
%w", nodeID, err)
+ return fmt.Errorf("failed to stream part to replica %d:
%w", replicaID, err)
}
}
@@ -629,8 +639,7 @@ func (mv *streamMigrationVisitor) streamPartToNode(nodeID
string, targetShardID
}
// Stream using chunked transfer (same as syncer.go:202)
- ctx := context.Background()
- result, err := chunkedClient.SyncStreamingParts(ctx,
[]queue.StreamingPartData{partData})
+ result, err := chunkedClient.SyncStreamingParts(mv.ctx,
[]queue.StreamingPartData{partData})
if err != nil {
return fmt.Errorf("failed to sync streaming parts to node %s:
%w", nodeID, err)
}
@@ -686,15 +695,22 @@ func (mv *streamMigrationVisitor) Close() error {
// createStreamingSegmentFromFiles creates StreamingPartData from segment
files.
func (mv *streamMigrationVisitor) createStreamingSegmentFromFiles(
targetShardID uint32,
- files []queue.FileInfo,
+ files []fileInfo,
segmentTR *timestamp.TimeRange,
topic string,
) queue.StreamingPartData {
+ filesInfo := make([]queue.FileInfo, 0, len(files))
+ for _, file := range files {
+ filesInfo = append(filesInfo, queue.FileInfo{
+ Name: file.name,
+ Reader: file.file.SequentialRead(),
+ })
+ }
segmentData := queue.StreamingPartData{
Group: mv.group,
ShardID: targetShardID, // Use calculated target shard
Topic: topic, // Use the new topic
- Files: files,
+ Files: filesInfo,
MinTimestamp: segmentTR.Start.UnixNano(),
MaxTimestamp: segmentTR.End.UnixNano(),
}
diff --git a/banyand/backup/lifecycle/trace_migration_visitor.go
b/banyand/backup/lifecycle/trace_migration_visitor.go
index 4c05dab01..2b5e2501e 100644
--- a/banyand/backup/lifecycle/trace_migration_visitor.go
+++ b/banyand/backup/lifecycle/trace_migration_visitor.go
@@ -43,6 +43,7 @@ import (
// traceMigrationVisitor implements the trace.Visitor interface for file-based
migration.
type traceMigrationVisitor struct {
+ ctx context.Context
client queue.Client
lfs fs.FileSystem
metadata metadata.Repo
@@ -64,11 +65,12 @@ type traceMigrationVisitor struct {
}
// newTraceMigrationVisitor creates a new file-based migration visitor.
-func newTraceMigrationVisitor(group *commonv1.Group, shardNum, replicas
uint32, selector node.Selector, client queue.Client,
+func newTraceMigrationVisitor(ctx context.Context, group *commonv1.Group,
shardNum, replicas uint32, selector node.Selector, client queue.Client,
l *logger.Logger, progress *Progress, chunkSize int,
targetStageInterval storage.IntervalRule, md metadata.Repo,
sourceStage, targetStage string, sourceSegmentInterval
storage.IntervalRule,
) *traceMigrationVisitor {
return &traceMigrationVisitor{
+ ctx: ctx,
group: group.Metadata.Name,
sourceStage: sourceStage,
targetStage: targetStage,
@@ -262,17 +264,11 @@ func (mv *traceMigrationVisitor) VisitSeries(segmentTR
*timestamp.TimeRange, ser
segmentIDStr := getSegmentTimeRange(targetSegmentTime,
mv.targetStageInterval).String()
for _, shardID := range shardIDs {
targetShardID :=
mv.calculateTargetShardID(uint32(shardID))
- ff := make([]queue.FileInfo, 0, len(files))
- for _, file := range files {
- ff = append(ff, queue.FileInfo{
- Name: file.name,
- Reader: file.file.SequentialRead(),
- })
- }
- partData :=
mv.createStreamingSegmentFromFiles(targetShardID, ff, segmentTR,
data.TopicTraceSeriesSync.String())
-
- // Stream segment to target shard replicas
- if err := mv.streamPartToTargetShard(targetShardID,
[]queue.StreamingPartData{partData}); err != nil {
+ // Stream segment to target shard replicas. The factory
rebuilds the
+ // part on every retry so each attempt gets fresh
offset-0 readers.
+ if err := mv.streamPartToTargetShard(targetShardID,
func() ([]queue.StreamingPartData, error) {
+ return
[]queue.StreamingPartData{mv.createStreamingSegmentFromFiles(targetShardID,
files, segmentTR, data.TopicTraceSeriesSync.String())}, nil
+ }); err != nil {
errorMsg := fmt.Sprintf("failed to stream trace
segment to target shard %d: %v", targetShardID, err)
mv.recordError(scopeSeries, segmentTR, shardID,
errorMsg)
return fmt.Errorf("failed to stream trace
segment to target shard %d: %w", targetShardID, err)
@@ -319,45 +315,59 @@ func (mv *traceMigrationVisitor) VisitShard(timestampTR
*timestamp.TimeRange, so
// Decide from the source shard's own [start, end) boundaries.
targetSegments := calculateTargetSegments(timestampTR.Start,
timestampTR.End, mv.targetStageInterval)
if len(targetSegments) > 1 {
- return mv.visitShardRowReplay(context.Background(),
timestampTR, sourceShardID, shardPath, segmentIDStr, targetSegments)
+ return mv.visitShardRowReplay(mv.ctx, timestampTR,
sourceShardID, shardPath, segmentIDStr, targetSegments)
}
atomic.AddUint64(&mv.partsCopiedSingleTarget, 1)
mv.progress.AddTraceChunkSyncShard(mv.group)
- allParts := make([]queue.StreamingPartData, 0)
- sidxPartData, sidxReleases, err :=
mv.generateAllSidxPartData(timestampTR, sourceShardID, filepath.Join(shardPath,
"sidx"))
- if err != nil {
- return fmt.Errorf("failed to generate sidx part data: %s: %w",
shardPath, err)
- }
- defer func() {
- for _, release := range sidxReleases {
- release()
- }
- }()
- allParts = append(allParts, sidxPartData...)
+ targetShardID := mv.calculateTargetShardID(uint32(sourceShardID))
- partDatas, partDataReleases, err := mv.generateAllPartData(timestampTR,
sourceShardID, shardPath)
- if err != nil {
- return fmt.Errorf("failed to generate core par data: %s: %w",
shardPath, err)
- }
+ // Regenerate the sidx + core parts each attempt for fresh offset-0
readers,
+ // releasing the prior attempt's handles first; the deferred call frees
the last set.
+ var prevRelease func()
defer func() {
- for _, release := range partDataReleases {
- release()
+ if prevRelease != nil {
+ prevRelease()
}
}()
- allParts = append(allParts, partDatas...)
-
- targetShardID := mv.calculateTargetShardID(uint32(sourceShardID))
+ mk := func() ([]queue.StreamingPartData, error) {
+ if prevRelease != nil {
+ prevRelease()
+ }
+ var releases []func()
+ // Captures releases by reference, so it frees whatever this
attempt has
+ // opened by call time — including the partial set on an early
return.
+ prevRelease = func() {
+ for _, release := range releases {
+ release()
+ }
+ }
+ allParts := make([]queue.StreamingPartData, 0)
+ sidxPartData, sidxReleases, genErr :=
mv.generateAllSidxPartData(timestampTR, sourceShardID, filepath.Join(shardPath,
"sidx"))
+ if genErr != nil {
+ return nil, fmt.Errorf("failed to generate sidx part
data: %s: %w", shardPath, genErr)
+ }
+ releases = append(releases, sidxReleases...)
+ allParts = append(allParts, sidxPartData...)
- sort.Slice(allParts, func(i, j int) bool {
- if allParts[i].ID == allParts[j].ID {
- return allParts[i].PartType < allParts[j].PartType
+ partDatas, partDataReleases, genErr :=
mv.generateAllPartData(timestampTR, sourceShardID, shardPath)
+ if genErr != nil {
+ return nil, fmt.Errorf("failed to generate core par
data: %s: %w", shardPath, genErr)
}
- return allParts[i].ID < allParts[j].ID
- })
+ releases = append(releases, partDataReleases...)
+ allParts = append(allParts, partDatas...)
+
+ sort.Slice(allParts, func(i, j int) bool {
+ if allParts[i].ID == allParts[j].ID {
+ return allParts[i].PartType <
allParts[j].PartType
+ }
+ return allParts[i].ID < allParts[j].ID
+ })
+ return allParts, nil
+ }
// Stream part to target segment
- if err := mv.streamPartToTargetShard(targetShardID, allParts); err !=
nil {
+ if err := mv.streamPartToTargetShard(targetShardID, mk); err != nil {
errorMsg := fmt.Errorf("failed to stream to target shard %d:
%w", targetShardID, err)
mv.recordError(scopeShard, timestampTR, sourceShardID,
errorMsg.Error())
return fmt.Errorf("failed to stream trace shard to target
segment shard: %w", err)
@@ -511,6 +521,12 @@ func (mv *traceMigrationVisitor) generateAllSidxPartData(
// Create StreamingPartData with PartType = index name
partData, err := sidx.ParsePartMetadata(mv.lfs,
partPath)
if err != nil {
+ // Release this part's just-opened handles and
all earlier parts'
+ // so a mid-way metadata failure does not leak
file descriptors.
+ release()
+ for _, r := range releases {
+ r()
+ }
return nil, nil, fmt.Errorf("failed to parse
sidx part metadata: %s: %w", partPath, err)
}
partData.Group = mv.group
@@ -564,6 +580,10 @@ func (mv *traceMigrationVisitor) generateAllPartData(
partPath := filepath.Join(shardPath, name)
parts, releases, err := mv.generatePartData(segmentTR,
sourceShardID, partPath)
if err != nil {
+ // Release earlier parts' handles so a mid-loop failure
does not leak.
+ for _, r := range allReleases {
+ r()
+ }
return nil, nil, fmt.Errorf("failed to generate part
data for path %s: %w", partPath, err)
}
allParts = append(allParts, parts...)
@@ -636,21 +656,24 @@ func (mv *traceMigrationVisitor)
calculateTargetShardID(sourceShardID uint32) ui
return calculateTargetShardID(sourceShardID, mv.targetShardNum)
}
-// streamPartToTargetShard sends part data to all replicas of the target shard.
-func (mv *traceMigrationVisitor) streamPartToTargetShard(targetShardID uint32,
partData []queue.StreamingPartData) error {
+// streamPartToTargetShard sends the parts to every replica with bounded
+// exponential-backoff retry (transient: target restarting, disconnect,
+// receiver SERVER_BUSY). streamPartToNode closes the parts' readers after each
+// send, so mk() is called per attempt to rebuild fresh offset-0 readers.
+func (mv *traceMigrationVisitor) streamPartToTargetShard(targetShardID uint32,
mk func() ([]queue.StreamingPartData, error)) error {
copies := mv.replicas + 1
// Send to all replicas using the exact pattern from steps.go:219-236
for replicaID := uint32(0); replicaID < copies; replicaID++ {
- // Use selector.Pick exactly like steps.go:220
- nodeID, err := mv.selector.Pick(mv.group, "", targetShardID,
replicaID)
+ err := pickAndRun(mv.ctx, mv.logger, mv.selector, mv.group, "",
targetShardID, replicaID, func(nodeID string) error {
+ partData, mkErr := mk()
+ if mkErr != nil {
+ return mkErr
+ }
+ return mv.streamPartToNode(nodeID, targetShardID,
partData)
+ })
if err != nil {
- return fmt.Errorf("failed to pick node for shard %d
replica %d: %w", targetShardID, replicaID, err)
- }
-
- // Stream part data to target node using chunked sync
- if err := mv.streamPartToNode(nodeID, targetShardID, partData);
err != nil {
- return fmt.Errorf("failed to stream trace part to node
%s: %w", nodeID, err)
+ return fmt.Errorf("failed to stream trace part to shard
%d replica %d: %w", targetShardID, replicaID, err)
}
}
@@ -672,8 +695,7 @@ func (mv *traceMigrationVisitor) streamPartToNode(nodeID
string, targetShardID u
}
// Stream using chunked transfer (same as syncer.go:202)
- ctx := context.Background()
- result, err := chunkedClient.SyncStreamingParts(ctx, partData)
+ result, err := chunkedClient.SyncStreamingParts(mv.ctx, partData)
if err != nil {
return fmt.Errorf("failed to sync streaming parts to node %s:
%w", nodeID, err)
}
@@ -726,15 +748,22 @@ func (mv *traceMigrationVisitor) Close() error {
// createStreamingSegmentFromFiles creates StreamingPartData from segment
files.
func (mv *traceMigrationVisitor) createStreamingSegmentFromFiles(
targetShardID uint32,
- files []queue.FileInfo,
+ files []fileInfo,
segmentTR *timestamp.TimeRange,
topic string,
) queue.StreamingPartData {
+ filesInfo := make([]queue.FileInfo, 0, len(files))
+ for _, file := range files {
+ filesInfo = append(filesInfo, queue.FileInfo{
+ Name: file.name,
+ Reader: file.file.SequentialRead(),
+ })
+ }
segmentData := queue.StreamingPartData{
Group: mv.group,
ShardID: targetShardID, // Use calculated target shard
Topic: topic, // Use the new topic
- Files: files,
+ Files: filesInfo,
MinTimestamp: segmentTR.Start.UnixNano(),
MaxTimestamp: segmentTR.End.UnixNano(),
}
diff --git a/banyand/measure/export_test.go b/banyand/measure/export_test.go
new file mode 100644
index 000000000..e0fbb9e5b
--- /dev/null
+++ b/banyand/measure/export_test.go
@@ -0,0 +1,37 @@
+// 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 measure
+
+import (
+ "time"
+
+ "github.com/apache/skywalking-banyandb/api/data"
+ "github.com/apache/skywalking-banyandb/banyand/protector"
+ "github.com/apache/skywalking-banyandb/banyand/queue"
+)
+
+// RegisterMeasureSeriesSyncHandlerForTest wires the REAL measure series-index
+// sync handler (setUpSyncSeriesCallback) of the supplied standalone service
onto
+// the supplied chunked-sync server, injecting the test-controlled protector
and
+// memWaitTimeout. It is the production registration from svc_data.go, exposed
to
+// the external measure_test package so an integration test can drive the
receive
+// path against a real target tsdb without spinning up the full data service.
+func RegisterMeasureSeriesSyncHandlerForTest(svc Service, server queue.Server,
pm protector.Memory, memWaitTimeout time.Duration) {
+ s := svc.(*standalone)
+ server.RegisterChunkedSyncHandler(data.TopicMeasureSeriesSync,
setUpSyncSeriesCallback(s.l, s.schemaRepo, pm, memWaitTimeout))
+}
diff --git a/banyand/measure/measure.go b/banyand/measure/measure.go
index 76a8c1ae3..ae18e38ef 100644
--- a/banyand/measure/measure.go
+++ b/banyand/measure/measure.go
@@ -56,6 +56,7 @@ type option struct {
seriesCacheMaxSize run.Bytes
flushTimeout time.Duration
syncInterval time.Duration
+ memWaitTimeout time.Duration
failedPartsMaxTotalSizeBytes uint64
vectorized vmeasure.VectorizedConfig
}
diff --git a/banyand/measure/migration_oom_throttling_test.go
b/banyand/measure/migration_oom_throttling_test.go
new file mode 100644
index 000000000..69ced1666
--- /dev/null
+++ b/banyand/measure/migration_oom_throttling_test.go
@@ -0,0 +1,408 @@
+// 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 measure_test
+
+import (
+ "context"
+ "errors"
+ "fmt"
+ "os"
+ "path/filepath"
+ "strings"
+ "sync/atomic"
+ "time"
+
+ . "github.com/onsi/ginkgo/v2"
+ . "github.com/onsi/gomega"
+ "github.com/spf13/cobra"
+ "google.golang.org/grpc"
+ "google.golang.org/grpc/credentials/insecure"
+
+ "github.com/apache/skywalking-banyandb/api/data"
+ commonv1
"github.com/apache/skywalking-banyandb/api/proto/banyandb/common/v1"
+ databasev1
"github.com/apache/skywalking-banyandb/api/proto/banyandb/database/v1"
+ "github.com/apache/skywalking-banyandb/banyand/internal/storage"
+ "github.com/apache/skywalking-banyandb/banyand/measure"
+ "github.com/apache/skywalking-banyandb/banyand/metadata/schema"
+ "github.com/apache/skywalking-banyandb/banyand/metadata/service"
+ "github.com/apache/skywalking-banyandb/banyand/observability"
+ obsservice
"github.com/apache/skywalking-banyandb/banyand/observability/services"
+ "github.com/apache/skywalking-banyandb/banyand/protector"
+ "github.com/apache/skywalking-banyandb/banyand/query"
+ "github.com/apache/skywalking-banyandb/banyand/queue"
+ "github.com/apache/skywalking-banyandb/banyand/queue/pub"
+ "github.com/apache/skywalking-banyandb/banyand/queue/sub"
+ "github.com/apache/skywalking-banyandb/pkg/fs"
+ "github.com/apache/skywalking-banyandb/pkg/run"
+ "github.com/apache/skywalking-banyandb/pkg/test"
+ "github.com/apache/skywalking-banyandb/pkg/test/helpers"
+)
+
+// controllablePM is a protector.Memory whose State() is flippable at runtime.
+// All other Memory methods
(AvailableBytes/GetLimit/AcquireResource/ShouldCache
+// + run.PreRunner/Config/Service) delegate to the embedded real protector so
the
+// receive handler behaves exactly like production except for the scripted
state.
+type controllablePM struct {
+ protector.Memory
+ high atomic.Bool
+}
+
+// State reports StateHigh while the test holds the protector high, StateLow
otherwise.
+func (c *controllablePM) State() protector.State {
+ if c.high.Load() {
+ return protector.StateHigh
+ }
+ return protector.StateLow
+}
+
+// migration OOM-throttling RECEIVE path, exercised end-to-end through a REAL
+// chunked-sync gRPC sender (pub) and receiver (sub) wired to the REAL measure
+// series-index sync handler (setUpSyncSeriesCallback) over a REAL target tsdb.
+// The ONLY fake is the protector: a controllablePM flipped between HIGH and
LOW.
+//
+// - While HIGH: syncSeriesCallback.HandleFileChunk returns
queue.ErrServerBusy
+// at chunk entry; the sub server translates that into
SYNC_STATUS_SERVER_BUSY,
+// which the pub client surfaces as an error wrapping queue.ErrServerBusy.
The
+// test asserts the migration of the series-index part is rejected as BUSY.
+// - After LOW: the same part migrates cleanly; the handler introduces the
+// external series-index segment (CompleteSegment) into the target tsdb.
The
+// test asserts (Eventually) the series index lands and the migrated tree
is
+// queryable, proving recovery.
+var _ = Describe("measure lifecycle tier-migration RECEIVE path under memory
pressure", func() {
+ const (
+ group = "oom_throttle_migration"
+ measureName = "oom_throttle_metric"
+ )
+
+ It("rejects the series-index part with SERVER_BUSY while memory is
HIGH, then completes once it drops to LOW", func() {
+ ctx := context.TODO()
+ twoDayInterval := storage.IntervalRule{Unit: storage.DAY, Num:
2}
+
+ workspace := GinkgoT().TempDir()
+ sourceRoot := filepath.Join(workspace, "source")
+ targetRoot := filepath.Join(workspace, "target")
+ Expect(os.MkdirAll(sourceRoot, 0o755)).To(Succeed())
+
+ // --- 1. Build a REAL source measure tree with a committed
sidx segment ---
+ srcSvcs, srcDeferFn := setUpMigrationTarget(sourceRoot)
+ sourceDown := func() {
+ if srcDeferFn != nil {
+ srcDeferFn()
+ srcDeferFn = nil
+ }
+ }
+ defer sourceDown()
+ Eventually(func() bool {
+ _, ok := srcSvcs.measure.LoadGroup("sw_metric")
+ return ok
+ }).WithTimeout(30 * time.Second).Should(BeTrue())
+ registerMigrationE2EGroup(srcSvcs, group, measureName)
+ Eventually(func() bool {
+ _, ok := srcSvcs.measure.LoadGroup(group)
+ return ok
+ }).WithTimeout(30 * time.Second).Should(BeTrue())
+ Eventually(func() error {
+ _, err :=
srcSvcs.measure.Measure(&commonv1.Metadata{Name: measureName, Group: group})
+ return err
+ }).WithTimeout(30*time.Second).Should(Succeed(),
+ "source service should resolve the measure before
writes are published")
+
+ // Two timestamps in a single DAY×2 bucket each, producing two
on-disk segs.
+ nowBucket := twoDayInterval.Standard(time.Now().UTC())
+ day1 := nowBucket.AddDate(0, 0, -4).Add(8 * time.Hour)
+ day2 := nowBucket.AddDate(0, 0, -2).Add(8 * time.Hour)
+ writeMigrationE2EPoints(srcSvcs, group, measureName, day1, 2, 0)
+ writeMigrationE2EPoints(srcSvcs, group, measureName, day2, 4, 2)
+
+ Eventually(func() int {
+ info, err := srcSvcs.measure.CollectDataInfo(ctx, group)
+ if err != nil || info == nil {
+ return 0
+ }
+ return len(info.SegmentInfo)
+
}).WithTimeout(60*time.Second).WithPolling(time.Second).Should(Equal(2),
+ "expected 2 DAY×2 segments on disk after writes are
flushed")
+
+ sourceDataPath := srcSvcs.measure.(interface{ GetDataPath()
string }).GetDataPath()
+ sourceGroupRoot := filepath.Join(sourceDataPath, group)
+ Eventually(func() bool {
+ return allSidxDirsHaveSnapshot(sourceGroupRoot)
+
}).WithTimeout(60*time.Second).WithPolling(time.Second).Should(BeTrue(),
+ "every <seg>/sidx/ under %s should carry a committed
.snp before migration runs", sourceGroupRoot)
+
+ // Stop the source so every per-segment bluge series-index
writer commits.
+ sourceDown()
+
+ // Collect the committed sidx .seg files (with their owning
segment dir) so
+ // we can replay the exact bytes the lifecycle sender ships.
+ segFiles := collectSidxSegFiles(sourceGroupRoot)
+ Expect(segFiles).NotTo(BeEmpty(), "source must have at least
one committed sidx .seg to migrate")
+ picked := segFiles[0]
+
+ // --- 2. REAL target measure receiver wired to a REAL sub gRPC
server ---
+ targetSvcs, targetPM, targetDeferFn :=
setUpThrottleTarget(targetRoot)
+ defer targetDeferFn()
+ Eventually(func() bool {
+ _, ok := targetSvcs.measure.LoadGroup("sw_metric")
+ return ok
+ }).WithTimeout(30 * time.Second).Should(BeTrue())
+ registerMigrationE2EGroup(targetSvcs, group, measureName)
+ Eventually(func() bool {
+ _, ok := targetSvcs.measure.LoadGroup(group)
+ return ok
+ }).WithTimeout(30 * time.Second).Should(BeTrue())
+ Eventually(func() error {
+ _, err :=
targetSvcs.measure.Measure(&commonv1.Metadata{Name: measureName, Group: group})
+ return err
+ }).WithTimeout(30*time.Second).Should(Succeed(),
+ "target service should resolve the measure before the
series part is received")
+
+ // Real chunked-sync server + the REAL series handler, fed the
fake protector.
+ // Keep memWaitTimeout small so the FinishSync WaitWhileHigh
path stays bounded.
+ const memWaitTimeout = 2 * time.Second
+ server, serverAddr, serverDeferFn := startThrottleSyncServer()
+ defer serverDeferFn()
+
measure.RegisterMeasureSeriesSyncHandlerForTest(targetSvcs.measure, server,
targetPM, memWaitTimeout)
+
+ // Real pub gRPC client; resolve the target node by address (no
metadata repo).
+ nodeName := "oom-throttle-target-node"
+ client := pub.NewWithoutMetadata(observability.BypassRegistry)
+ client.OnAddOrUpdate(schema.Metadata{
+ TypeMeta: schema.TypeMeta{Name: nodeName, Kind:
schema.KindNode},
+ Spec: &databasev1.Node{
+ Metadata: &commonv1.Metadata{Name: nodeName},
+ Roles:
[]databasev1.Role{databasev1.Role_ROLE_DATA},
+ GrpcAddress: serverAddr,
+ },
+ })
+ var chunkedClient queue.ChunkedSyncClient
+ Eventually(func() error {
+ var clientErr error
+ chunkedClient, clientErr =
client.NewChunkedSyncClient(nodeName, 1024)
+ return clientErr
+ }).WithTimeout(30 * time.Second).Should(Succeed())
+ defer func() {
+ _ = chunkedClient.Close()
+ }()
+
+ // mkSeriesPart rebuilds fresh offset-0 readers for the picked
sidx .seg, so
+ // each migration attempt streams the same bytes (the sender
reopens on retry).
+ lfs := fs.NewLocalFileSystem()
+ mkSeriesPart := func() queue.StreamingPartData {
+ f, openErr := lfs.OpenFile(picked.path)
+ Expect(openErr).NotTo(HaveOccurred())
+ return queue.StreamingPartData{
+ Group: group,
+ ShardID: 0,
+ Topic:
data.TopicMeasureSeriesSync.String(),
+ Files: []queue.FileInfo{{Name:
picked.name, Reader: f.SequentialRead()}},
+ MinTimestamp: picked.minTS,
+ MaxTimestamp: picked.maxTS,
+ }
+ }
+
+ // --- 3. HIGH → assert SERVER_BUSY is observed on the receive
path ---
+ targetPM.high.Store(true)
+ _, busyErr := chunkedClient.SyncStreamingParts(ctx,
[]queue.StreamingPartData{mkSeriesPart()})
+ Expect(busyErr).To(HaveOccurred(),
+ "while memory is HIGH the receiver must shed the series
part instead of accepting it")
+ Expect(errors.Is(busyErr, queue.ErrServerBusy)).To(BeTrue(),
+ "the receive-side SYNC_STATUS_SERVER_BUSY must surface
to the sender as queue.ErrServerBusy, got: %v", busyErr)
+
+ // The target must not have introduced the series index while
busy.
+ Consistently(func() bool {
+ return targetSidxHasSnapshot(filepath.Join(targetRoot,
"measure", "data", group))
+
}).WithTimeout(time.Second).WithPolling(200*time.Millisecond).Should(BeFalse(),
+ "no series-index segment should be introduced into the
target while memory is HIGH")
+
+ // --- 4. Drop to LOW (normal) ---
+ targetPM.high.Store(false)
+
+ // --- 5. Eventually the migration COMPLETES and the series
index lands ---
+ Eventually(func() error {
+ result, syncErr :=
chunkedClient.SyncStreamingParts(ctx, []queue.StreamingPartData{mkSeriesPart()})
+ if syncErr != nil {
+ return fmt.Errorf("sync still failing after
LOW: %w", syncErr)
+ }
+ if !result.Success {
+ return fmt.Errorf("sync result not successful:
failedParts=%v", result.FailedParts)
+ }
+ return nil
+
}).WithTimeout(30*time.Second).WithPolling(500*time.Millisecond).Should(Succeed(),
+ "once memory recovers to LOW the series-index part must
migrate cleanly")
+
+ // The receiver introduced the external series-index segment
into the target.
+ Eventually(func() bool {
+ return targetSidxHasSnapshot(filepath.Join(targetRoot,
"measure", "data", group))
+
}).WithTimeout(30*time.Second).WithPolling(500*time.Millisecond).Should(BeTrue(),
+ "after recovery the target must carry a committed
series-index snapshot for the migrated segment")
+ })
+})
+
+// sidxSegFile locates one committed sidx .seg file under the source tree and
the
+// time range of its owning segment, so the test can frame the
StreamingPartData
+// exactly as the lifecycle sender's createStreamingSegmentFromFiles does.
+type sidxSegFile struct {
+ path string
+ name string
+ minTS int64
+ maxTS int64
+}
+
+// collectSidxSegFiles walks <groupRoot>/seg-*/sidx and returns every *.seg
file
+// with the [start, end) of its owning segment (parsed from the seg dir name on
+// the DAY×2 grid), which the receive handler uses to route the external
segment.
+func collectSidxSegFiles(groupRoot string) []sidxSegFile {
+ twoDayInterval := storage.IntervalRule{Unit: storage.DAY, Num: 2}
+ segs, err := os.ReadDir(groupRoot)
+ Expect(err).NotTo(HaveOccurred())
+ var out []sidxSegFile
+ for _, seg := range segs {
+ if !seg.IsDir() || !strings.HasPrefix(seg.Name(), "seg-") {
+ continue
+ }
+ start, parseErr := parseSegDirStart(seg.Name())
+ if parseErr != nil {
+ continue
+ }
+ end := twoDayInterval.NextTime(start)
+ sidxDir := filepath.Join(groupRoot, seg.Name(), "sidx")
+ entries, readErr := os.ReadDir(sidxDir)
+ if readErr != nil {
+ continue
+ }
+ for _, e := range entries {
+ if e.IsDir() || !strings.HasSuffix(e.Name(), ".seg") {
+ continue
+ }
+ out = append(out, sidxSegFile{
+ path: filepath.Join(sidxDir, e.Name()),
+ name: e.Name(),
+ minTS: start.UnixNano(),
+ maxTS: end.UnixNano() - 1,
+ })
+ }
+ }
+ return out
+}
+
+// parseSegDirStart parses a "seg-YYYYMMDD" directory name into its start
instant.
+func parseSegDirStart(name string) (time.Time, error) {
+ const segPrefix = "seg-"
+ raw := strings.TrimPrefix(name, segPrefix)
+ return time.ParseInLocation("20060102", raw, time.Local)
+}
+
+// targetSidxHasSnapshot reports whether any <groupRoot>/seg-*/sidx carries a
+// committed .seg file (the external segment introduced by CompleteSegment).
+func targetSidxHasSnapshot(groupRoot string) bool {
+ segs, err := os.ReadDir(groupRoot)
+ if err != nil {
+ return false
+ }
+ for _, seg := range segs {
+ if !seg.IsDir() || !strings.HasPrefix(seg.Name(), "seg-") {
+ continue
+ }
+ sidxDir := filepath.Join(groupRoot, seg.Name(), "sidx")
+ entries, readErr := os.ReadDir(sidxDir)
+ if readErr != nil {
+ continue
+ }
+ for _, e := range entries {
+ if !e.IsDir() && strings.HasSuffix(e.Name(), ".seg") {
+ return true
+ }
+ }
+ }
+ return false
+}
+
+// setUpThrottleTarget brings up a REAL standalone measure receiver whose
memory
+// protector is a controllablePM (so the test can flip HIGH/LOW), returning the
+// services bundle and the protector handle.
+func setUpThrottleTarget(measureRootPath string) (*services, *controllablePM,
func()) {
+ pipeline := queue.Local()
+ metadataService, err := service.NewService()
+ Expect(err).NotTo(HaveOccurred())
+ metricSvc := obsservice.NewMetricService(metadataService, pipeline,
"test", nil)
+ pm := &controllablePM{Memory: protector.NewMemory(metricSvc)}
+ measureService, err := measure.NewStandalone(metadataService, pipeline,
nil, metricSvc, pm)
+ Expect(err).NotTo(HaveOccurred())
+ preloadSvc := &preloadMeasureService{metaSvc: metadataService}
+ querySvc, err := query.NewService(context.TODO(), nil, measureService,
nil, metadataService, pipeline, metricSvc, false)
+ Expect(err).NotTo(HaveOccurred())
+
+ metaPath, metaDeferFunc, err := test.NewSpace()
+ Expect(err).NotTo(HaveOccurred())
+ schemaPorts, err := test.AllocateFreePorts(1)
+ Expect(err).NotTo(HaveOccurred())
+ flags := []string{
+ "--schema-server-root-path=" + metaPath,
+ fmt.Sprintf("--schema-server-grpc-port=%d", schemaPorts[0]),
+ "--schema-server-grpc-host=127.0.0.1",
+ "--measure-root-path=" + measureRootPath,
+ }
+ moduleDeferFunc := test.SetupModules(flags, pipeline, metadataService,
preloadSvc, measureService, querySvc)
+ return &services{
+ measure: measureService,
+ metadataService: metadataService,
+ pipeline: pipeline,
+ }, pm, func() {
+ moduleDeferFunc()
+ metaDeferFunc()
+ }
+}
+
+// startThrottleSyncServer starts a REAL chunked-sync gRPC server and returns
it
+// alongside its localhost address. It mirrors the proven setup in
+// banyand/queue/test: a run.Group driven by a cobra command so the server's
+// flags are registered and parsed before Run, otherwise the listener never
binds.
+// The series handler is attached by the caller via
RegisterMeasureSeriesSyncHandlerForTest.
+func startThrottleSyncServer() (queue.Server, string, func()) {
+ ports, err := test.AllocateFreePorts(2)
+ Expect(err).NotTo(HaveOccurred())
+ grpcPort := uint32(ports[0])
+ httpPort := uint32(ports[1])
+
+ server := sub.NewServerWithPorts(observability.BypassRegistry,
"oom-throttle-sync", grpcPort, httpPort)
+ closer, deferFn := run.NewTester("oom-throttle-sync-closer")
+ g := run.NewGroup("oom-throttle-sync")
+ g.Register(closer, server)
+
+ cmd := &cobra.Command{
+ Use: "oom-throttle-sync",
+ FParseErrWhitelist: cobra.FParseErrWhitelist{UnknownFlags:
true},
+ Run: func(_ *cobra.Command, _ []string) {
+ Expect(g.Run(context.Background())).To(Succeed())
+ },
+ }
+ cmd.Flags().AddFlagSet(g.RegisterFlags().FlagSet)
+ go func() {
+ defer GinkgoRecover()
+ Expect(cmd.Execute()).To(Succeed())
+ }()
+
+ addr := fmt.Sprintf("localhost:%d", grpcPort)
+ Eventually(func() error {
+ return helpers.HealthCheck(addr, 10*time.Second, 10*time.Second,
+
grpc.WithTransportCredentials(insecure.NewCredentials()))()
+ }).WithTimeout(30 * time.Second).Should(Succeed())
+
+ return server, addr, deferFn
+}
diff --git a/banyand/measure/svc_data.go b/banyand/measure/svc_data.go
index 24aeeb4f1..f69c19f2b 100644
--- a/banyand/measure/svc_data.go
+++ b/banyand/measure/svc_data.go
@@ -170,6 +170,8 @@ func (s *dataSVC) FlagSet() *run.FlagSet {
flagS.StringVar(&s.root, "measure-root-path", "/tmp", "the root path of
measure")
flagS.StringVar(&s.dataPath, "measure-data-path", "", "the data
directory path of measure. If not set, <measure-root-path>/measure/data will be
used")
flagS.DurationVar(&s.option.flushTimeout, "measure-flush-timeout",
defaultFlushTimeout, "the memory data timeout of measure")
+ flagS.DurationVar(&s.option.memWaitTimeout,
"measure-lifecycle-receive-mem-wait-timeout", 5*time.Minute,
+ "max time the migration receiver waits for memory to recover
before introducing an external segment")
s.option.mergePolicy = newDefaultMergePolicy()
flagS.VarP(&s.option.mergePolicy.maxFanOutSize,
"measure-max-fan-out-size", "", "the upper bound of a single file size after
merge of measure")
s.option.seriesCacheMaxSize = run.Bytes(32 << 20)
@@ -310,7 +312,7 @@ func (s *dataSVC) PreRun(ctx context.Context) error {
}
s.pipeline.RegisterChunkedSyncHandler(data.TopicMeasurePartSync,
setUpChunkedSyncCallback(s.l, s.schemaRepo))
- s.pipeline.RegisterChunkedSyncHandler(data.TopicMeasureSeriesSync,
setUpSyncSeriesCallback(s.l, s.schemaRepo))
+ s.pipeline.RegisterChunkedSyncHandler(data.TopicMeasureSeriesSync,
setUpSyncSeriesCallback(s.l, s.schemaRepo, s.pm, s.option.memWaitTimeout))
err = s.pipeline.Subscribe(data.TopicMeasureSeriesIndexInsert,
setUpIndexCallback(s.l, s.schemaRepo, data.TopicMeasureSeriesIndexInsert))
if err != nil {
return err
diff --git a/banyand/measure/svc_liaison.go b/banyand/measure/svc_liaison.go
index d34752ed9..3e5449d8b 100644
--- a/banyand/measure/svc_liaison.go
+++ b/banyand/measure/svc_liaison.go
@@ -23,6 +23,7 @@ import (
"path"
"path/filepath"
"strings"
+ "time"
"github.com/pkg/errors"
@@ -109,6 +110,8 @@ func (s *liaison) FlagSet() *run.FlagSet {
flagS.StringVar(&s.root, "measure-root-path", "/tmp", "the root path of
measure")
flagS.StringVar(&s.dataPath, "measure-data-path", "", "the data
directory path of measure. If not set, <measure-root-path>/measure/data will be
used")
flagS.DurationVar(&s.option.flushTimeout, "measure-flush-timeout",
defaultFlushTimeout, "the memory data timeout of measure")
+ flagS.DurationVar(&s.option.memWaitTimeout,
"measure-lifecycle-receive-mem-wait-timeout", 5*time.Minute,
+ "max time the migration receiver waits for memory to recover
before introducing an external segment")
flagS.DurationVar(&s.option.syncInterval, "measure-sync-interval",
defaultSyncInterval, "the periodic sync interval for measure data")
flagS.IntVar(&s.maxDiskUsagePercent, "measure-max-disk-usage-percent",
95, "the maximum disk usage percentage allowed")
flagS.IntVar(&s.failedPartsMaxSizePercent,
"failed-parts-max-size-percent", 10,
diff --git a/banyand/measure/svc_standalone.go
b/banyand/measure/svc_standalone.go
index 9e5eb6358..8be02a81c 100644
--- a/banyand/measure/svc_standalone.go
+++ b/banyand/measure/svc_standalone.go
@@ -177,6 +177,8 @@ func (s *standalone) FlagSet() *run.FlagSet {
flagS.StringVar(&s.root, "measure-root-path", "/tmp", "the root path of
measure")
flagS.StringVar(&s.dataPath, "measure-data-path", "", "the data
directory path of measure. If not set, <measure-root-path>/measure/data will be
used")
flagS.DurationVar(&s.option.flushTimeout, "measure-flush-timeout",
defaultFlushTimeout, "the memory data timeout of measure")
+ flagS.DurationVar(&s.option.memWaitTimeout,
"measure-lifecycle-receive-mem-wait-timeout", 5*time.Minute,
+ "max time the migration receiver waits for memory to recover
before introducing an external segment")
s.option.mergePolicy = newDefaultMergePolicy()
flagS.VarP(&s.option.mergePolicy.maxFanOutSize,
"measure-max-fan-out-size", "", "the upper bound of a single file size after
merge of measure")
s.option.seriesCacheMaxSize = run.Bytes(32 << 20)
diff --git a/banyand/measure/write_data.go b/banyand/measure/write_data.go
index c6079f365..301449a82 100644
--- a/banyand/measure/write_data.go
+++ b/banyand/measure/write_data.go
@@ -18,6 +18,7 @@
package measure
import (
+ "context"
"fmt"
"strings"
"sync/atomic"
@@ -26,6 +27,7 @@ import (
"github.com/apache/skywalking-banyandb/api/common"
commonv1
"github.com/apache/skywalking-banyandb/api/proto/banyandb/common/v1"
"github.com/apache/skywalking-banyandb/banyand/internal/storage"
+ "github.com/apache/skywalking-banyandb/banyand/protector"
"github.com/apache/skywalking-banyandb/banyand/queue"
"github.com/apache/skywalking-banyandb/pkg/fs"
"github.com/apache/skywalking-banyandb/pkg/index"
@@ -158,6 +160,9 @@ func (s *syncCallback) HandleFileChunk(ctx
*queue.ChunkedSyncPartContext, chunk
return fmt.Errorf("part handler is nil")
}
partCtx := ctx.Handler.(*syncPartContext)
+ if partCtx.tsTable.pm.State() == protector.StateHigh {
+ return queue.ErrServerBusy
+ }
// Select the appropriate writer based on the filename and write the
chunk.
fileName := ctx.FileName
@@ -214,10 +219,13 @@ func releaseWriters(sw *writers) {
var writersPool = pool.Register[*writers]("measure-writers")
type syncSeriesContext struct {
- streamer index.ExternalSegmentStreamer
- segment storage.Segment[*tsTable, *commonv1.ResourceOpts]
- l *logger.Logger
- fileName string
+ streamer index.ExternalSegmentStreamer
+ segment storage.Segment[*tsTable, *commonv1.ResourceOpts]
+ l *logger.Logger
+ pm protector.Memory
+ introduceCtx context.Context
+ fileName string
+ memWaitTimeout time.Duration
}
func (s *syncSeriesContext) NewPartType(_ *queue.ChunkedSyncPartContext) error
{
@@ -227,6 +235,9 @@ func (s *syncSeriesContext) NewPartType(_
*queue.ChunkedSyncPartContext) error {
func (s *syncSeriesContext) FinishSync() error {
if s.streamer != nil {
+ if err := protector.WaitWhileHigh(s.introduceCtx, s.pm,
s.memWaitTimeout, s.l, "measure-series-introduce"); err != nil {
+ return err
+ }
if err := s.streamer.CompleteSegment(); err != nil {
s.l.Error().Err(err).Msg("failed to complete external
segment")
return err
@@ -246,14 +257,18 @@ func (s *syncSeriesContext) Close() error {
}
type syncSeriesCallback struct {
- l *logger.Logger
- schemaRepo *schemaRepo
+ l *logger.Logger
+ schemaRepo *schemaRepo
+ pm protector.Memory
+ memWaitTimeout time.Duration
}
-func setUpSyncSeriesCallback(l *logger.Logger, schemaRepo *schemaRepo)
queue.ChunkedSyncHandler {
+func setUpSyncSeriesCallback(l *logger.Logger, schemaRepo *schemaRepo, pm
protector.Memory, memWaitTimeout time.Duration) queue.ChunkedSyncHandler {
return &syncSeriesCallback{
- l: l,
- schemaRepo: schemaRepo,
+ l: l,
+ schemaRepo: schemaRepo,
+ pm: pm,
+ memWaitTimeout: memWaitTimeout,
}
}
@@ -278,8 +293,11 @@ func (s *syncSeriesCallback) CreatePartHandler(ctx
*queue.ChunkedSyncPartContext
return nil, err
}
return &syncSeriesContext{
- l: s.l,
- segment: segment,
+ l: s.l,
+ segment: segment,
+ pm: s.pm,
+ memWaitTimeout: s.memWaitTimeout,
+ introduceCtx: ctx.RetrieveContext(),
}, nil
}
@@ -289,12 +307,18 @@ func (s *syncSeriesCallback) HandleFileChunk(ctx
*queue.ChunkedSyncPartContext,
return fmt.Errorf("part handler is nil")
}
seriesCtx := ctx.Handler.(*syncSeriesContext)
+ if seriesCtx.pm.State() == protector.StateHigh {
+ return queue.ErrServerBusy
+ }
if seriesCtx.segment == nil {
return fmt.Errorf("segment is nil")
}
if seriesCtx.fileName != ctx.FileName {
if seriesCtx.streamer != nil {
+ if err :=
protector.WaitWhileHigh(ctx.RetrieveContext(), seriesCtx.pm,
seriesCtx.memWaitTimeout, s.l, "measure-series-introduce"); err != nil {
+ return err
+ }
if err := seriesCtx.streamer.CompleteSegment(); err !=
nil {
s.l.Error().Err(err).Str("group",
ctx.Group).Msg("failed to complete external segment")
return err
diff --git a/banyand/measure/write_data_busy_test.go
b/banyand/measure/write_data_busy_test.go
new file mode 100644
index 000000000..7ffb1bdf4
--- /dev/null
+++ b/banyand/measure/write_data_busy_test.go
@@ -0,0 +1,67 @@
+// 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 measure
+
+import (
+ "testing"
+
+ "github.com/stretchr/testify/assert"
+ "github.com/stretchr/testify/require"
+
+ "github.com/apache/skywalking-banyandb/banyand/protector"
+ "github.com/apache/skywalking-banyandb/banyand/queue"
+ "github.com/apache/skywalking-banyandb/pkg/logger"
+)
+
+// highMem is a fake protector.Memory that always reports a configurable state.
+type highMem struct {
+ protector.Nop
+ state protector.State
+}
+
+// State returns the fixed scripted state.
+func (h highMem) State() protector.State {
+ return h.state
+}
+
+// TestSyncSeriesCallback_HandleFileChunkBusy verifies that the series chunk
+// handler sheds load by returning queue.ErrServerBusy when memory is high,
+// before touching any segment or streamer state.
+func TestSyncSeriesCallback_HandleFileChunkBusy(t *testing.T) {
+ cb := &syncSeriesCallback{l: logger.GetLogger("test")}
+ // seriesCtx has a nil segment; the BUSY check must fire first so the
+ // nil-segment path is never reached.
+ seriesCtx := &syncSeriesContext{
+ l: logger.GetLogger("test"),
+ pm: highMem{state: protector.StateHigh},
+ }
+ ctx := &queue.ChunkedSyncPartContext{Handler: seriesCtx, FileName:
"series.idx"}
+ err := cb.HandleFileChunk(ctx, []byte("payload"))
+ require.Error(t, err)
+ assert.ErrorIs(t, err, queue.ErrServerBusy)
+}
+
+// TestSyncSeriesCallback_HandleFileChunkNilHandler verifies the early nil
+// handler guard still applies.
+func TestSyncSeriesCallback_HandleFileChunkNilHandler(t *testing.T) {
+ cb := &syncSeriesCallback{l: logger.GetLogger("test")}
+ ctx := &queue.ChunkedSyncPartContext{}
+ err := cb.HandleFileChunk(ctx, []byte("payload"))
+ require.Error(t, err)
+ assert.NotErrorIs(t, err, queue.ErrServerBusy)
+}
diff --git a/banyand/protector/wait.go b/banyand/protector/wait.go
new file mode 100644
index 000000000..5d8357448
--- /dev/null
+++ b/banyand/protector/wait.go
@@ -0,0 +1,63 @@
+// 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 protector
+
+import (
+ "context"
+ "time"
+
+ "github.com/apache/skywalking-banyandb/pkg/logger"
+)
+
+const (
+ waitInitialBackoff = 200 * time.Millisecond
+ waitMaxBackoff = 2 * time.Second
+ waitHeartbeat = time.Minute
+)
+
+// WaitWhileHigh blocks while memory pressure is high, bounded by maxWait.
+// It only creates a timeout context when it actually has to wait. It returns
+// the timeout/cancel error if ctx or maxWait elapses before recovery.
+func WaitWhileHigh(ctx context.Context, pm Memory, maxWait time.Duration, l
*logger.Logger, stage string) error {
+ if pm.State() != StateHigh {
+ return nil
+ }
+ waitCtx, cancel := context.WithTimeout(ctx, maxWait)
+ defer cancel()
+ start := time.Now()
+ l.Warn().Str("stage", stage).Dur("max_wait", maxWait).Msg("memory high,
migration receive paused, waiting")
+ ticker := time.NewTicker(waitHeartbeat)
+ defer ticker.Stop()
+ backoff := waitInitialBackoff
+ for pm.State() == StateHigh {
+ select {
+ case <-time.After(backoff):
+ backoff *= 2
+ if backoff > waitMaxBackoff {
+ backoff = waitMaxBackoff
+ }
+ case <-ticker.C:
+ l.Warn().Str("stage", stage).Dur("waited",
time.Since(start)).Msg("still waiting for memory")
+ case <-waitCtx.Done():
+ l.Warn().Str("stage", stage).Dur("waited",
time.Since(start)).Msg("memory wait aborted (timeout/cancel)")
+ return waitCtx.Err()
+ }
+ }
+ l.Info().Str("stage", stage).Dur("waited",
time.Since(start)).Msg("memory recovered, migration receive resumed")
+ return nil
+}
diff --git a/banyand/protector/wait_test.go b/banyand/protector/wait_test.go
new file mode 100644
index 000000000..eec29127e
--- /dev/null
+++ b/banyand/protector/wait_test.go
@@ -0,0 +1,72 @@
+// 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 protector
+
+import (
+ "context"
+ "sync/atomic"
+ "testing"
+ "time"
+
+ "github.com/stretchr/testify/assert"
+ "github.com/stretchr/testify/require"
+
+ "github.com/apache/skywalking-banyandb/pkg/logger"
+)
+
+// fakeMem is a test double that implements the full Memory interface by
+// embedding Nop and overriding State() to advance through a script of states.
+type fakeMem struct {
+ Nop
+ states []State
+ idx atomic.Int32
+}
+
+// State returns the next scripted state, sticking on the last value.
+func (f *fakeMem) State() State {
+ i := int(f.idx.Load())
+ if i >= len(f.states) {
+ i = len(f.states) - 1
+ } else {
+ f.idx.Add(1)
+ }
+ return f.states[i]
+}
+
+func TestWaitWhileHigh_ReturnsWhenRecovered(t *testing.T) {
+ pm := &fakeMem{states: []State{StateHigh, StateHigh, StateLow}}
+ err := WaitWhileHigh(context.Background(), pm, time.Second,
logger.GetLogger("test"), "test-stage")
+ assert.NoError(t, err)
+}
+
+func TestWaitWhileHigh_TimesOut(t *testing.T) {
+ pm := &fakeMem{states: []State{StateHigh}}
+ start := time.Now()
+ err := WaitWhileHigh(context.Background(), pm, 50*time.Millisecond,
logger.GetLogger("test"), "test-stage")
+ require.Error(t, err)
+ assert.ErrorIs(t, err, context.DeadlineExceeded)
+ assert.Less(t, time.Since(start), time.Second, "timeout test must
complete quickly")
+}
+
+func TestWaitWhileHigh_FastPathNoTimeoutCtx(t *testing.T) {
+ pm := &fakeMem{states: []State{StateLow}}
+ ctx, cancel := context.WithCancel(context.Background())
+ defer cancel()
+ err := WaitWhileHigh(ctx, pm, time.Second, logger.GetLogger("test"),
"test-stage")
+ assert.NoError(t, err)
+}
diff --git a/banyand/queue/pub/chunked_sync.go
b/banyand/queue/pub/chunked_sync.go
index 6c30fef0b..cc77e2a68 100644
--- a/banyand/queue/pub/chunked_sync.go
+++ b/banyand/queue/pub/chunked_sync.go
@@ -512,6 +512,9 @@ func (c *chunkedSyncClient) sendChunk(
case clusterv1.SyncStatus_SYNC_STATUS_SESSION_NOT_FOUND:
return fmt.Errorf("session %s not found on server for
chunk %d: %s", sessionID, *chunkIndex, resp.Error)
+ case clusterv1.SyncStatus_SYNC_STATUS_SERVER_BUSY:
+ return fmt.Errorf("receiver busy for chunk %d: %w",
*chunkIndex, queue.ErrServerBusy)
+
default:
if resp.Error != "" {
return fmt.Errorf("chunk %d sync failed: %s",
*chunkIndex, resp.Error)
@@ -572,6 +575,9 @@ func classifyChunkedSyncPubErr(err error) string {
if err == nil {
return ""
}
+ if errors.Is(err, queue.ErrServerBusy) {
+ return "server_busy"
+ }
msg := err.Error()
switch {
case strings.Contains(msg, "checksum mismatch"):
diff --git a/banyand/queue/queue.go b/banyand/queue/queue.go
index 62944bd49..fc0c03d61 100644
--- a/banyand/queue/queue.go
+++ b/banyand/queue/queue.go
@@ -40,6 +40,10 @@ import (
// self-probing without a special-cased branch in the caller.
var ErrNotImplemented = errors.New("not implemented")
+// ErrServerBusy signals the receiver is under memory pressure and the sender
+// should back off and retry the whole part.
+var ErrServerBusy = errors.New("receiver under memory pressure")
+
// Queue builds a data transmission tunnel between subscribers and publishers.
//
//go:generate mockgen -destination=./queue_mock.go -package=queue
github.com/apache/skywalking-banyandb/pkg/bus MessageListener
@@ -136,7 +140,10 @@ type ChunkedSyncClient interface {
// ChunkedSyncPartContext represents the context for a chunked sync operation.
type ChunkedSyncPartContext struct {
- Handler PartHandler
+ Handler PartHandler
+ // Context carries the gRPC stream context for receive-side cancellation
+ // (memory wait upper bound / disconnect).
+ Context context.Context
Group string
FileName string
PartType string
@@ -152,6 +159,15 @@ type ChunkedSyncPartContext struct {
ShardID uint32
}
+// RetrieveContext returns the part's stream context, or context.Background()
+// if none was set. Used by receive handlers to bound memory-pressure waits.
+func (c *ChunkedSyncPartContext) RetrieveContext() context.Context {
+ if c.Context == nil {
+ return context.Background()
+ }
+ return c.Context
+}
+
// Close releases resources associated with the context.
func (c *ChunkedSyncPartContext) Close() error {
defer func() {
diff --git a/banyand/queue/queue_test.go b/banyand/queue/queue_test.go
new file mode 100644
index 000000000..8030d9211
--- /dev/null
+++ b/banyand/queue/queue_test.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 queue
+
+import (
+ "context"
+ "testing"
+
+ "github.com/stretchr/testify/assert"
+)
+
+// TestChunkedSyncPartContextRetrieveContext verifies the nil fallback and
+// pass-through behavior of ChunkedSyncPartContext.RetrieveContext.
+func TestChunkedSyncPartContextRetrieveContext(t *testing.T) {
+ assert.Equal(t, context.Background(),
(&ChunkedSyncPartContext{}).RetrieveContext())
+
+ ctx, cancel := context.WithCancel(context.Background())
+ defer cancel()
+ assert.Equal(t, ctx, (&ChunkedSyncPartContext{Context:
ctx}).RetrieveContext())
+}
diff --git a/banyand/queue/sub/chunked_sync.go
b/banyand/queue/sub/chunked_sync.go
index 1ed1be1fa..dedca8e3f 100644
--- a/banyand/queue/sub/chunked_sync.go
+++ b/banyand/queue/sub/chunked_sync.go
@@ -418,6 +418,7 @@ func (s *server) processExpectedChunk(stream
clusterv1.ChunkedSyncService_SyncPa
if createNewContext {
session.partCtx = &queue.ChunkedSyncPartContext{
+ Context: stream.Context(),
ID: partInfo.Id,
Group: session.metadata.Group,
ShardID: session.metadata.ShardId,
@@ -452,6 +453,10 @@ func (s *server) processExpectedChunk(stream
clusterv1.ChunkedSyncService_SyncPa
}
if processErr := s.processPart(session, req, partInfo,
partIndex, handler); processErr != nil {
+ if errors.Is(processErr, queue.ErrServerBusy) {
+ return s.sendResponse(stream, req,
clusterv1.SyncStatus_SYNC_STATUS_SERVER_BUSY,
+ "receiver under memory pressure, retry
later", nil)
+ }
s.log.Error().Err(processErr).
Str("session_id", req.SessionId).
Str("topic", session.metadata.Topic).
diff --git a/banyand/queue/sub/chunked_sync_test.go
b/banyand/queue/sub/chunked_sync_test.go
index 206f93b8c..f43c970bb 100644
--- a/banyand/queue/sub/chunked_sync_test.go
+++ b/banyand/queue/sub/chunked_sync_test.go
@@ -282,6 +282,68 @@ func TestChunkOrderingBufferFull(t *testing.T) {
assert.True(t, found, "expected error_type=buffer_full to be recorded")
}
+// TestChunkedSyncServerBusy verifies that when the handler's HandleFileChunk
reports
+// memory pressure via queue.ErrServerBusy, the server replies with
SYNC_STATUS_SERVER_BUSY
+// instead of a generic stream error or a CHUNK_RECEIVED success.
+func TestChunkedSyncServerBusy(t *testing.T) {
+ err := logger.Init(logger.Logging{Env: "dev", Level: "info"})
+ require.NoError(t, err)
+
+ s := &server{ //nolint:exhaustruct
+ log: logger.GetLogger("test-server-busy"),
+ chunkedSyncHandlers:
make(map[bus.Topic]queue.ChunkedSyncHandler),
+ }
+ s.chunkedSyncHandlers[data.TopicStreamPartSync] =
&busyChunkedSyncHandler{}
+
+ mockStream := &MockSyncPartStream{}
+ session := &syncSession{ //nolint:exhaustruct
+ sessionID: "busy-session",
+ startTime: time.Now(),
+ partsProgress: make(map[int]*partProgress),
+ metadata: &clusterv1.SyncMetadata{
+ Group: "test-group",
+ ShardId: 1,
+ Topic: data.TopicStreamPartSync.String(),
+ },
+ }
+
+ chunkData := []byte("busy-chunk-data")
+ req := &clusterv1.SyncPartRequest{
+ SessionId: "busy-session",
+ ChunkIndex: 0,
+ ChunkData: chunkData,
+ ChunkChecksum: fmt.Sprintf("%x", crc32.ChecksumIEEE(chunkData)),
+ PartsInfo: []*clusterv1.PartInfo{
+ {
+ Id: 1,
+ Files: []*clusterv1.FileInfo{
+ {Name: "test-file.dat", Offset: 0,
Size: uint32(len(chunkData))},
+ },
+ },
+ },
+ }
+
+ require.NoError(t, s.processChunk(mockStream, session, req))
+
+ require.Len(t, mockStream.sentResponses, 1)
+ resp := mockStream.sentResponses[0]
+ assert.Equal(t, clusterv1.SyncStatus_SYNC_STATUS_SERVER_BUSY,
resp.Status)
+ assert.NotEmpty(t, resp.Error)
+ assert.Equal(t, "busy-session", resp.SessionId)
+ assert.Equal(t, uint32(0), resp.ChunkIndex)
+}
+
+// busyChunkedSyncHandler is a handler whose HandleFileChunk always reports
memory pressure.
+type busyChunkedSyncHandler struct{}
+
+func (m *busyChunkedSyncHandler) CreatePartHandler(_
*queue.ChunkedSyncPartContext) (queue.PartHandler, error) {
+ return &MockChunkedSyncPartHandler{}, nil
+}
+
+func (m *busyChunkedSyncHandler) HandleFileChunk(_
*queue.ChunkedSyncPartContext, _ []byte) error {
+ return queue.ErrServerBusy
+}
+
// MockChunkedSyncHandler implements queue.ChunkedSyncHandler for testing.
type MockChunkedSyncHandler struct{}
diff --git a/banyand/stream/stream.go b/banyand/stream/stream.go
index a5b3873ac..7d47871f1 100644
--- a/banyand/stream/stream.go
+++ b/banyand/stream/stream.go
@@ -54,6 +54,7 @@ type option struct {
flushTimeout time.Duration
elementIndexFlushTimeout time.Duration
syncInterval time.Duration
+ memWaitTimeout time.Duration
failedPartsMaxTotalSizeBytes uint64
}
diff --git a/banyand/stream/svc_liaison.go b/banyand/stream/svc_liaison.go
index 78da92c9c..176b7f6ba 100644
--- a/banyand/stream/svc_liaison.go
+++ b/banyand/stream/svc_liaison.go
@@ -23,6 +23,7 @@ import (
"path"
"path/filepath"
"strings"
+ "time"
"github.com/pkg/errors"
@@ -116,6 +117,8 @@ func (s *liaison) FlagSet() *run.FlagSet {
flagS.StringVar(&s.root, "stream-root-path", "/tmp", "the root path of
stream")
flagS.StringVar(&s.dataPath, "stream-data-path", "", "the data
directory path of stream. If not set, <stream-root-path>/stream/data will be
used")
flagS.DurationVar(&s.option.flushTimeout, "stream-flush-timeout",
defaultFlushTimeout, "the memory data timeout of stream")
+ flagS.DurationVar(&s.option.memWaitTimeout,
"stream-lifecycle-receive-mem-wait-timeout", 5*time.Minute,
+ "max time the migration receiver waits for memory to recover
before introducing an external segment")
flagS.IntVar(&s.maxDiskUsagePercent, "stream-max-disk-usage-percent",
95, "the maximum disk usage percentage allowed")
flagS.DurationVar(&s.option.syncInterval, "stream-sync-interval",
defaultSyncInterval, "the periodic sync interval for stream data")
flagS.IntVar(&s.failedPartsMaxSizePercent,
"failed-parts-max-size-percent", 10,
diff --git a/banyand/stream/svc_standalone.go b/banyand/stream/svc_standalone.go
index 9dab8a867..86172aa1d 100644
--- a/banyand/stream/svc_standalone.go
+++ b/banyand/stream/svc_standalone.go
@@ -177,6 +177,8 @@ func (s *standalone) FlagSet() *run.FlagSet {
flagS.StringVar(&s.dataPath, "stream-data-path", "", "the data
directory path of stream. If not set, <stream-root-path>/stream/data will be
used")
flagS.DurationVar(&s.option.flushTimeout, "stream-flush-timeout",
defaultFlushTimeout, "the memory data timeout of stream")
flagS.DurationVar(&s.option.elementIndexFlushTimeout,
"element-index-flush-timeout", defaultFlushTimeout, "the elementIndex timeout
of stream")
+ flagS.DurationVar(&s.option.memWaitTimeout,
"stream-lifecycle-receive-mem-wait-timeout", 5*time.Minute,
+ "max time the migration receiver waits for memory to recover
before introducing an external segment")
s.option.mergePolicy = newDefaultMergePolicy()
flagS.VarP(&s.option.mergePolicy.maxFanOutSize,
"stream-max-fan-out-size", "", "the upper bound of a single file size after
merge of stream")
s.option.seriesCacheMaxSize = run.Bytes(32 << 20)
@@ -293,9 +295,9 @@ func (s *standalone) PreRun(ctx context.Context) error {
}
s.pipeline.RegisterChunkedSyncHandler(data.TopicStreamPartSync,
setUpChunkedSyncCallback(s.l, &s.schemaRepo))
// Register chunked sync handler for stream series index
- s.pipeline.RegisterChunkedSyncHandler(data.TopicStreamSeriesSync,
setUpSyncSeriesCallback(s.l, &s.schemaRepo))
+ s.pipeline.RegisterChunkedSyncHandler(data.TopicStreamSeriesSync,
setUpSyncSeriesCallback(s.l, &s.schemaRepo, s.pm, s.option.memWaitTimeout))
// Register chunked sync handler for stream element index
- s.pipeline.RegisterChunkedSyncHandler(data.TopicStreamElementIndexSync,
setUpSyncElementIndexCallback(s.l, &s.schemaRepo))
+ s.pipeline.RegisterChunkedSyncHandler(data.TopicStreamElementIndexSync,
setUpSyncElementIndexCallback(s.l, &s.schemaRepo, s.pm,
s.option.memWaitTimeout))
err = s.pipeline.Subscribe(data.TopicStreamSeriesIndexWrite,
setUpSeriesIndexCallback(s.l, &s.schemaRepo))
if err != nil {
diff --git a/banyand/stream/write_data.go b/banyand/stream/write_data.go
index 8506a36e7..af360c9c4 100644
--- a/banyand/stream/write_data.go
+++ b/banyand/stream/write_data.go
@@ -18,6 +18,7 @@
package stream
import (
+ "context"
"fmt"
"strings"
"sync/atomic"
@@ -26,6 +27,7 @@ import (
"github.com/apache/skywalking-banyandb/api/common"
commonv1
"github.com/apache/skywalking-banyandb/api/proto/banyandb/common/v1"
"github.com/apache/skywalking-banyandb/banyand/internal/storage"
+ "github.com/apache/skywalking-banyandb/banyand/protector"
"github.com/apache/skywalking-banyandb/banyand/queue"
"github.com/apache/skywalking-banyandb/pkg/fs"
"github.com/apache/skywalking-banyandb/pkg/index"
@@ -157,6 +159,9 @@ func (s *syncCallback) HandleFileChunk(ctx
*queue.ChunkedSyncPartContext, chunk
return fmt.Errorf("part handler is nil")
}
partCtx := ctx.Handler.(*syncPartContext)
+ if partCtx.tsTable.pm.State() == protector.StateHigh {
+ return queue.ErrServerBusy
+ }
// Select the appropriate writer based on the filename and write the
chunk.
fileName := ctx.FileName
@@ -188,10 +193,13 @@ func (s *syncCallback) HandleFileChunk(ctx
*queue.ChunkedSyncPartContext, chunk
}
type syncSeriesContext struct {
- streamer index.ExternalSegmentStreamer
- segment storage.Segment[*tsTable, *commonv1.ResourceOpts]
- l *logger.Logger
- fileName string
+ streamer index.ExternalSegmentStreamer
+ segment storage.Segment[*tsTable, *commonv1.ResourceOpts]
+ l *logger.Logger
+ pm protector.Memory
+ introduceCtx context.Context
+ fileName string
+ memWaitTimeout time.Duration
}
func (s *syncSeriesContext) NewPartType(_ *queue.ChunkedSyncPartContext) error
{
@@ -201,6 +209,9 @@ func (s *syncSeriesContext) NewPartType(_
*queue.ChunkedSyncPartContext) error {
func (s *syncSeriesContext) FinishSync() error {
if s.streamer != nil {
+ if err := protector.WaitWhileHigh(s.introduceCtx, s.pm,
s.memWaitTimeout, s.l, "stream-series-introduce"); err != nil {
+ return err
+ }
if err := s.streamer.CompleteSegment(); err != nil {
s.l.Error().Err(err).Msg("failed to complete external
segment")
return err
@@ -220,14 +231,18 @@ func (s *syncSeriesContext) Close() error {
}
type syncSeriesCallback struct {
- l *logger.Logger
- schemaRepo *schemaRepo
+ l *logger.Logger
+ schemaRepo *schemaRepo
+ pm protector.Memory
+ memWaitTimeout time.Duration
}
-func setUpSyncSeriesCallback(l *logger.Logger, schemaRepo *schemaRepo)
queue.ChunkedSyncHandler {
+func setUpSyncSeriesCallback(l *logger.Logger, schemaRepo *schemaRepo, pm
protector.Memory, memWaitTimeout time.Duration) queue.ChunkedSyncHandler {
return &syncSeriesCallback{
- l: l,
- schemaRepo: schemaRepo,
+ l: l,
+ schemaRepo: schemaRepo,
+ pm: pm,
+ memWaitTimeout: memWaitTimeout,
}
}
@@ -252,8 +267,11 @@ func (s *syncSeriesCallback) CreatePartHandler(ctx
*queue.ChunkedSyncPartContext
return nil, err
}
return &syncSeriesContext{
- l: s.l,
- segment: segment,
+ l: s.l,
+ segment: segment,
+ pm: s.pm,
+ memWaitTimeout: s.memWaitTimeout,
+ introduceCtx: ctx.RetrieveContext(),
}, nil
}
@@ -263,12 +281,18 @@ func (s *syncSeriesCallback) HandleFileChunk(ctx
*queue.ChunkedSyncPartContext,
return fmt.Errorf("part handler is nil")
}
seriesCtx := ctx.Handler.(*syncSeriesContext)
+ if seriesCtx.pm.State() == protector.StateHigh {
+ return queue.ErrServerBusy
+ }
if seriesCtx.segment == nil {
return fmt.Errorf("segment is nil")
}
if seriesCtx.fileName != ctx.FileName {
if seriesCtx.streamer != nil {
+ if err :=
protector.WaitWhileHigh(ctx.RetrieveContext(), seriesCtx.pm,
seriesCtx.memWaitTimeout, s.l, "stream-series-introduce"); err != nil {
+ return err
+ }
if err := seriesCtx.streamer.CompleteSegment(); err !=
nil {
s.l.Error().Err(err).Str("group",
ctx.Group).Msg("failed to complete external segment")
return err
@@ -294,11 +318,14 @@ func (s *syncSeriesCallback) HandleFileChunk(ctx
*queue.ChunkedSyncPartContext,
}
type syncElementIndexContext struct {
- streamer index.ExternalSegmentStreamer
- segment storage.Segment[*tsTable, *commonv1.ResourceOpts]
- tsTable *tsTable
- l *logger.Logger
- fileName string
+ streamer index.ExternalSegmentStreamer
+ segment storage.Segment[*tsTable, *commonv1.ResourceOpts]
+ tsTable *tsTable
+ l *logger.Logger
+ pm protector.Memory
+ introduceCtx context.Context
+ fileName string
+ memWaitTimeout time.Duration
}
func (s *syncElementIndexContext) NewPartType(_ *queue.ChunkedSyncPartContext)
error {
@@ -308,6 +335,9 @@ func (s *syncElementIndexContext) NewPartType(_
*queue.ChunkedSyncPartContext) e
func (s *syncElementIndexContext) FinishSync() error {
if s.streamer != nil {
+ if err := protector.WaitWhileHigh(s.introduceCtx, s.pm,
s.memWaitTimeout, s.l, "stream-element-index-introduce"); err != nil {
+ return err
+ }
if err := s.streamer.CompleteSegment(); err != nil {
s.l.Error().Err(err).Msg("failed to complete external
segment for element index")
return err
@@ -328,14 +358,18 @@ func (s *syncElementIndexContext) Close() error {
}
type syncElementIndexCallback struct {
- l *logger.Logger
- schemaRepo *schemaRepo
+ l *logger.Logger
+ schemaRepo *schemaRepo
+ pm protector.Memory
+ memWaitTimeout time.Duration
}
-func setUpSyncElementIndexCallback(l *logger.Logger, schemaRepo *schemaRepo)
queue.ChunkedSyncHandler {
+func setUpSyncElementIndexCallback(l *logger.Logger, schemaRepo *schemaRepo,
pm protector.Memory, memWaitTimeout time.Duration) queue.ChunkedSyncHandler {
return &syncElementIndexCallback{
- l: l,
- schemaRepo: schemaRepo,
+ l: l,
+ schemaRepo: schemaRepo,
+ pm: pm,
+ memWaitTimeout: memWaitTimeout,
}
}
@@ -366,9 +400,12 @@ func (s *syncElementIndexCallback) CreatePartHandler(ctx
*queue.ChunkedSyncPartC
}
return &syncElementIndexContext{
- l: s.l,
- tsTable: tsTable,
- segment: segment,
+ l: s.l,
+ tsTable: tsTable,
+ segment: segment,
+ pm: s.pm,
+ memWaitTimeout: s.memWaitTimeout,
+ introduceCtx: ctx.RetrieveContext(),
}, nil
}
@@ -378,12 +415,18 @@ func (s *syncElementIndexCallback) HandleFileChunk(ctx
*queue.ChunkedSyncPartCon
return fmt.Errorf("part handler is nil")
}
elementCtx := ctx.Handler.(*syncElementIndexContext)
+ if elementCtx.pm.State() == protector.StateHigh {
+ return queue.ErrServerBusy
+ }
if elementCtx.tsTable == nil {
return fmt.Errorf("ts table is nil")
}
if elementCtx.fileName != ctx.FileName {
if elementCtx.streamer != nil {
+ if err :=
protector.WaitWhileHigh(ctx.RetrieveContext(), elementCtx.pm,
elementCtx.memWaitTimeout, s.l, "stream-element-index-introduce"); err != nil {
+ return err
+ }
if err := elementCtx.streamer.CompleteSegment(); err !=
nil {
s.l.Error().Err(err).Str("group",
ctx.Group).Msg("failed to complete external segment for element index")
return err
diff --git a/banyand/stream/write_data_busy_test.go
b/banyand/stream/write_data_busy_test.go
new file mode 100644
index 000000000..de05c7068
--- /dev/null
+++ b/banyand/stream/write_data_busy_test.go
@@ -0,0 +1,94 @@
+// 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 stream
+
+import (
+ "testing"
+
+ "github.com/stretchr/testify/assert"
+ "github.com/stretchr/testify/require"
+
+ "github.com/apache/skywalking-banyandb/banyand/protector"
+ "github.com/apache/skywalking-banyandb/banyand/queue"
+ "github.com/apache/skywalking-banyandb/pkg/logger"
+)
+
+// highMem is a fake protector.Memory that always reports a configurable state.
+type highMem struct {
+ protector.Nop
+ state protector.State
+}
+
+// State returns the fixed scripted state.
+func (h highMem) State() protector.State {
+ return h.state
+}
+
+// TestSyncSeriesCallback_HandleFileChunkBusy verifies that the series chunk
+// handler sheds load by returning queue.ErrServerBusy when memory is high,
+// before touching any segment or streamer state.
+func TestSyncSeriesCallback_HandleFileChunkBusy(t *testing.T) {
+ cb := &syncSeriesCallback{l: logger.GetLogger("test")}
+ // seriesCtx has a nil segment; the BUSY check must fire first so the
+ // nil-segment path is never reached.
+ seriesCtx := &syncSeriesContext{
+ l: logger.GetLogger("test"),
+ pm: highMem{state: protector.StateHigh},
+ }
+ ctx := &queue.ChunkedSyncPartContext{Handler: seriesCtx, FileName:
"series.idx"}
+ err := cb.HandleFileChunk(ctx, []byte("payload"))
+ require.Error(t, err)
+ assert.ErrorIs(t, err, queue.ErrServerBusy)
+}
+
+// TestSyncSeriesCallback_HandleFileChunkNilHandler verifies the early nil
+// handler guard still applies.
+func TestSyncSeriesCallback_HandleFileChunkNilHandler(t *testing.T) {
+ cb := &syncSeriesCallback{l: logger.GetLogger("test")}
+ ctx := &queue.ChunkedSyncPartContext{}
+ err := cb.HandleFileChunk(ctx, []byte("payload"))
+ require.Error(t, err)
+ assert.NotErrorIs(t, err, queue.ErrServerBusy)
+}
+
+// TestSyncElementIndexCallback_HandleFileChunkBusy verifies that the element
+// index chunk handler sheds load by returning queue.ErrServerBusy when memory
+// is high, before touching any ts table or streamer state.
+func TestSyncElementIndexCallback_HandleFileChunkBusy(t *testing.T) {
+ cb := &syncElementIndexCallback{l: logger.GetLogger("test")}
+ // elementCtx has a nil tsTable; the BUSY check must fire first so the
+ // nil-tsTable path is never reached.
+ elementCtx := &syncElementIndexContext{
+ l: logger.GetLogger("test"),
+ pm: highMem{state: protector.StateHigh},
+ }
+ ctx := &queue.ChunkedSyncPartContext{Handler: elementCtx, FileName:
"element.idx"}
+ err := cb.HandleFileChunk(ctx, []byte("payload"))
+ require.Error(t, err)
+ assert.ErrorIs(t, err, queue.ErrServerBusy)
+}
+
+// TestSyncElementIndexCallback_HandleFileChunkNilHandler verifies the early
nil
+// handler guard still applies.
+func TestSyncElementIndexCallback_HandleFileChunkNilHandler(t *testing.T) {
+ cb := &syncElementIndexCallback{l: logger.GetLogger("test")}
+ ctx := &queue.ChunkedSyncPartContext{}
+ err := cb.HandleFileChunk(ctx, []byte("payload"))
+ require.Error(t, err)
+ assert.NotErrorIs(t, err, queue.ErrServerBusy)
+}
diff --git a/banyand/trace/svc_liaison.go b/banyand/trace/svc_liaison.go
index 92a2418f1..053c9c657 100644
--- a/banyand/trace/svc_liaison.go
+++ b/banyand/trace/svc_liaison.go
@@ -129,6 +129,8 @@ func (l *liaison) FlagSet() *run.FlagSet {
fs.StringVar(&l.root, "trace-root-path", "/tmp", "the root path for
trace data")
fs.StringVar(&l.dataPath, "trace-data-path", "", "the path for trace
data (optional)")
fs.DurationVar(&l.option.flushTimeout, "trace-flush-timeout",
3*time.Second, "the timeout for trace data flush")
+ fs.DurationVar(&l.option.memWaitTimeout,
"trace-lifecycle-receive-mem-wait-timeout", 5*time.Minute,
+ "max time the migration receiver waits for memory to recover
before introducing an external segment")
fs.IntVar(&l.maxDiskUsagePercent, "trace-max-disk-usage-percent", 95,
"the maximum disk usage percentage")
fs.DurationVar(&l.option.syncInterval, "trace-sync-interval",
defaultSyncInterval, "the periodic sync interval for trace data")
fs.StringSliceVar(&l.dataNodeList, "data-node-list", nil,
"comma-separated list of data node names to monitor for handoff")
diff --git a/banyand/trace/svc_standalone.go b/banyand/trace/svc_standalone.go
index 4f0f657a6..a2df01723 100644
--- a/banyand/trace/svc_standalone.go
+++ b/banyand/trace/svc_standalone.go
@@ -83,6 +83,8 @@ func (s *standalone) FlagSet() *run.FlagSet {
fs.StringVar(&s.root, "trace-root-path", "/tmp", "the root path for
trace data")
fs.StringVar(&s.dataPath, "trace-data-path", "", "the path for trace
data (optional)")
fs.DurationVar(&s.option.flushTimeout, "trace-flush-timeout",
defaultFlushTimeout, "the timeout for trace data flush")
+ fs.DurationVar(&s.option.memWaitTimeout,
"trace-lifecycle-receive-mem-wait-timeout", 5*time.Minute,
+ "max time the migration receiver waits for memory to recover
before introducing an external segment")
// Retention configuration flags
fs.Float64Var(&s.retentionConfig.HighWatermark,
"trace-retention-high-watermark", 95.0, "disk usage high watermark percentage
that triggers forced retention cleanup")
@@ -200,7 +202,7 @@ func (s *standalone) PreRun(ctx context.Context) error {
return err
}
s.pipeline.RegisterChunkedSyncHandler(data.TopicTracePartSync,
setUpChunkedSyncCallback(s.l, &s.schemaRepo))
- s.pipeline.RegisterChunkedSyncHandler(data.TopicTraceSeriesSync,
setUpSeriesSyncCallback(s.l, &s.schemaRepo))
+ s.pipeline.RegisterChunkedSyncHandler(data.TopicTraceSeriesSync,
setUpSeriesSyncCallback(s.l, &s.schemaRepo, s.pm, s.option.memWaitTimeout))
err = s.pipeline.Subscribe(data.TopicTraceSidxSeriesWrite,
setUpSidxSeriesIndexCallback(s.l, &s.schemaRepo))
if err != nil {
return err
diff --git a/banyand/trace/trace.go b/banyand/trace/trace.go
index 884972740..d27a324e4 100644
--- a/banyand/trace/trace.go
+++ b/banyand/trace/trace.go
@@ -58,6 +58,7 @@ type option struct {
seriesCacheMaxSize run.Bytes
flushTimeout time.Duration
syncInterval time.Duration
+ memWaitTimeout time.Duration
failedPartsMaxTotalSizeBytes uint64
vectorized vtrace.VectorizedConfig
}
diff --git a/banyand/trace/write_data.go b/banyand/trace/write_data.go
index 7a57ba4ce..e28155235 100644
--- a/banyand/trace/write_data.go
+++ b/banyand/trace/write_data.go
@@ -18,6 +18,7 @@
package trace
import (
+ "context"
"fmt"
"path/filepath"
"strings"
@@ -28,6 +29,7 @@ import (
commonv1
"github.com/apache/skywalking-banyandb/api/proto/banyandb/common/v1"
"github.com/apache/skywalking-banyandb/banyand/internal/sidx"
"github.com/apache/skywalking-banyandb/banyand/internal/storage"
+ "github.com/apache/skywalking-banyandb/banyand/protector"
"github.com/apache/skywalking-banyandb/banyand/queue"
"github.com/apache/skywalking-banyandb/pkg/fs"
"github.com/apache/skywalking-banyandb/pkg/index"
@@ -158,10 +160,13 @@ func (s *syncPartContext) Close() error {
}
type syncSeriesContext struct {
- streamer index.ExternalSegmentStreamer
- segment storage.Segment[*tsTable, *commonv1.ResourceOpts]
- l *logger.Logger
- fileName string
+ streamer index.ExternalSegmentStreamer
+ segment storage.Segment[*tsTable, *commonv1.ResourceOpts]
+ l *logger.Logger
+ pm protector.Memory
+ introduceCtx context.Context
+ fileName string
+ memWaitTimeout time.Duration
}
func (s *syncSeriesContext) NewPartType(_ *queue.ChunkedSyncPartContext) error
{
@@ -171,6 +176,9 @@ func (s *syncSeriesContext) NewPartType(_
*queue.ChunkedSyncPartContext) error {
func (s *syncSeriesContext) FinishSync() error {
if s.streamer != nil {
+ if err := protector.WaitWhileHigh(s.introduceCtx, s.pm,
s.memWaitTimeout, s.l, "trace-series-introduce"); err != nil {
+ return err
+ }
if err := s.streamer.CompleteSegment(); err != nil {
s.l.Error().Err(err).Msg("failed to complete external
segment")
return err
@@ -190,14 +198,18 @@ func (s *syncSeriesContext) Close() error {
}
type syncSeriesCallback struct {
- l *logger.Logger
- schemaRepo *schemaRepo
+ l *logger.Logger
+ schemaRepo *schemaRepo
+ pm protector.Memory
+ memWaitTimeout time.Duration
}
-func setUpSeriesSyncCallback(l *logger.Logger, s *schemaRepo)
queue.ChunkedSyncHandler {
+func setUpSeriesSyncCallback(l *logger.Logger, s *schemaRepo, pm
protector.Memory, memWaitTimeout time.Duration) queue.ChunkedSyncHandler {
return &syncSeriesCallback{
- l: l,
- schemaRepo: s,
+ l: l,
+ schemaRepo: s,
+ pm: pm,
+ memWaitTimeout: memWaitTimeout,
}
}
@@ -221,8 +233,11 @@ func (s *syncSeriesCallback) CreatePartHandler(ctx
*queue.ChunkedSyncPartContext
return nil, err
}
return &syncSeriesContext{
- l: s.l,
- segment: segment,
+ l: s.l,
+ segment: segment,
+ pm: s.pm,
+ memWaitTimeout: s.memWaitTimeout,
+ introduceCtx: ctx.RetrieveContext(),
}, nil
}
@@ -232,12 +247,18 @@ func (s *syncSeriesCallback) HandleFileChunk(ctx
*queue.ChunkedSyncPartContext,
return fmt.Errorf("part handler is nil")
}
seriesCtx := ctx.Handler.(*syncSeriesContext)
+ if seriesCtx.pm.State() == protector.StateHigh {
+ return queue.ErrServerBusy
+ }
if seriesCtx.segment == nil {
return fmt.Errorf("segment is nil")
}
if seriesCtx.fileName != ctx.FileName {
if seriesCtx.streamer != nil {
+ if err :=
protector.WaitWhileHigh(ctx.RetrieveContext(), seriesCtx.pm,
seriesCtx.memWaitTimeout, s.l, "trace-series-introduce"); err != nil {
+ return err
+ }
if err := seriesCtx.streamer.CompleteSegment(); err !=
nil {
s.l.Error().Err(err).Str("group",
ctx.Group).Msg("failed to complete external segment")
return err
@@ -322,6 +343,10 @@ func (s *syncChunkCallback) HandleFileChunk(ctx
*queue.ChunkedSyncPartContext, c
if ctx.Handler == nil {
return fmt.Errorf("part handler is nil")
}
+ partCtx := ctx.Handler.(*syncPartContext)
+ if partCtx.tsTable.pm.State() == protector.StateHigh {
+ return queue.ErrServerBusy
+ }
if ctx.PartType != PartTypeCore {
return s.handleSidxFileChunk(ctx, chunk)
}
diff --git a/banyand/trace/write_data_busy_test.go
b/banyand/trace/write_data_busy_test.go
new file mode 100644
index 000000000..837db3a3b
--- /dev/null
+++ b/banyand/trace/write_data_busy_test.go
@@ -0,0 +1,67 @@
+// 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 (
+ "testing"
+
+ "github.com/stretchr/testify/assert"
+ "github.com/stretchr/testify/require"
+
+ "github.com/apache/skywalking-banyandb/banyand/protector"
+ "github.com/apache/skywalking-banyandb/banyand/queue"
+ "github.com/apache/skywalking-banyandb/pkg/logger"
+)
+
+// highMem is a fake protector.Memory that always reports a configurable state.
+type highMem struct {
+ protector.Nop
+ state protector.State
+}
+
+// State returns the fixed scripted state.
+func (h highMem) State() protector.State {
+ return h.state
+}
+
+// TestSyncSeriesCallback_HandleFileChunkBusy verifies that the series chunk
+// handler sheds load by returning queue.ErrServerBusy when memory is high,
+// before touching any segment or streamer state.
+func TestSyncSeriesCallback_HandleFileChunkBusy(t *testing.T) {
+ cb := &syncSeriesCallback{l: logger.GetLogger("test")}
+ // seriesCtx has a nil segment; the BUSY check must fire first so the
+ // nil-segment path is never reached.
+ seriesCtx := &syncSeriesContext{
+ l: logger.GetLogger("test"),
+ pm: highMem{state: protector.StateHigh},
+ }
+ ctx := &queue.ChunkedSyncPartContext{Handler: seriesCtx, FileName:
"series.idx"}
+ err := cb.HandleFileChunk(ctx, []byte("payload"))
+ require.Error(t, err)
+ assert.ErrorIs(t, err, queue.ErrServerBusy)
+}
+
+// TestSyncSeriesCallback_HandleFileChunkNilHandler verifies the early nil
+// handler guard still applies.
+func TestSyncSeriesCallback_HandleFileChunkNilHandler(t *testing.T) {
+ cb := &syncSeriesCallback{l: logger.GetLogger("test")}
+ ctx := &queue.ChunkedSyncPartContext{}
+ err := cb.HandleFileChunk(ctx, []byte("payload"))
+ require.Error(t, err)
+ assert.NotErrorIs(t, err, queue.ErrServerBusy)
+}
diff --git a/dist/LICENSE b/dist/LICENSE
index 6c74d9919..1d66f2beb 100644
--- a/dist/LICENSE
+++ b/dist/LICENSE
@@ -189,7 +189,7 @@ Apache-2.0 licenses
github.com/GoogleCloudPlatform/opentelemetry-operations-go/detectors/gcp
v1.32.0 Apache-2.0
github.com/GoogleCloudPlatform/opentelemetry-operations-go/exporter/metric
v0.56.0 Apache-2.0
github.com/GoogleCloudPlatform/opentelemetry-operations-go/internal/resourcemapping
v0.56.0 Apache-2.0
- github.com/SkyAPM/bluge v0.0.0-20250804100126-cccf29a55f01 Apache-2.0
+ github.com/SkyAPM/bluge v0.0.0-20260625022800-42385daf66b8 Apache-2.0
github.com/SkyAPM/ice v0.0.0-20250619023539-b5173603b0b3 Apache-2.0
github.com/SkyAPM/ktm-ebpf v0.0.0-20260228024820-81a19d950bff Apache-2.0
github.com/apache/skywalking-cli v0.0.0-20240227151024-ee371a210afe
Apache-2.0
diff --git a/docs/api-reference.md b/docs/api-reference.md
index 4026f3260..e9a706dfa 100644
--- a/docs/api-reference.md
+++ b/docs/api-reference.md
@@ -2693,6 +2693,7 @@ SyncStatus represents the status of a sync operation.
| SYNC_STATUS_SYNC_COMPLETE | 5 | Entire sync operation completed
successfully. |
| SYNC_STATUS_VERSION_UNSUPPORTED | 6 | Version not supported for sync
operations. |
| SYNC_STATUS_FORMAT_VERSION_MISMATCH | 7 | File format version incompatible. |
+| SYNC_STATUS_SERVER_BUSY | 8 | Receiver under memory pressure; sender should
back off and retry the whole part. |
diff --git a/docs/operation/configuration.md b/docs/operation/configuration.md
index 6b6a39890..599b34c07 100644
--- a/docs/operation/configuration.md
+++ b/docs/operation/configuration.md
@@ -121,6 +121,7 @@ The following flags are used to configure the measure
storage engine:
- `--measure-root-path string`: The root path of the measure database
(default: "/tmp").
- `--measure-data-path string`: The data directory path of measure. If not
set, `<measure-root-path>/measure/data` is used.
- `--measure-max-fan-out-size bytes`: the upper bound of a single file size
after merge of measure (default 8.00EiB)
+- `--measure-lifecycle-receive-mem-wait-timeout duration`: Max time a
lifecycle-migration receiver waits for memory to recover before introducing an
external segment (default: 5m).
The following flags are used to configure the stream storage engine:
@@ -129,12 +130,14 @@ The following flags are used to configure the stream
storage engine:
- `--stream-data-path string`: The data directory path of stream. If not set,
`<stream-root-path>/stream/data` is used.
- `--stream-max-fan-out-size bytes`: the upper bound of a single file size
after merge of stream (default 8.00EiB)
- `--element-index-flush-timeout duration`: The element index timeout of
stream (default: 1s).
+- `--stream-lifecycle-receive-mem-wait-timeout duration`: Max time a
lifecycle-migration receiver waits for memory to recover before introducing an
external segment (default: 5m).
The following flags are used to configure the trace storage engine:
- `--trace-flush-timeout duration`: The memory data timeout of trace (default:
1s).
- `--trace-root-path string`: The root path of the database (default: "/tmp").
- `--trace-max-fan-out-size bytes`: the upper bound of a single file size
after merge of trace (default 8.00EiB)
+- `--trace-lifecycle-receive-mem-wait-timeout duration`: Max time a
lifecycle-migration receiver waits for memory to recover before introducing an
external segment (default: 5m).
The following flags configure the remaining per-catalog storage roots:
diff --git a/docs/operation/lifecycle.md b/docs/operation/lifecycle.md
index a04153a27..7a1d66566 100644
--- a/docs/operation/lifecycle.md
+++ b/docs/operation/lifecycle.md
@@ -135,6 +135,7 @@ lifecycle \
| `--schedule` | Schedule for periodic backup (e.g.,
@yearly, @monthly, @weekly, @daily, etc.) | `""`
|
| `--migration-orphan-policy` | What to do with rows whose
measure/stream schema was deleted from the registry: `archive` or `discard` |
`archive` |
| `--migration-orphan-archive-subdir` | Relative subdirectory, under each
catalog's root path, for archived orphan rows when policy is `archive` |
`archive` |
+| `--lifecycle-send-retry-timeout` | Maximum total time to retry streaming
a migration part to a target node on transient failures (target node
restarting, disconnect, receiver busy) | `15m` |
## Handling Orphan (Deleted-Schema) Data
diff --git a/go.mod b/go.mod
index e4fe08b6a..70af9b785 100644
--- a/go.mod
+++ b/go.mod
@@ -208,7 +208,7 @@ require (
replace (
github.com/benbjohnson/clock v1.3.0 => github.com/SkyAPM/clock
v1.3.1-0.20220809233656-dc7607c94a97
- github.com/blugelabs/bluge => github.com/SkyAPM/bluge
v0.0.0-20250804100126-cccf29a55f01
+ github.com/blugelabs/bluge => github.com/SkyAPM/bluge
v0.0.0-20260625022800-42385daf66b8
github.com/blugelabs/bluge_segment_api =>
github.com/zinclabs/bluge_segment_api v1.0.0
github.com/blugelabs/ice => github.com/SkyAPM/ice
v0.0.0-20250619023539-b5173603b0b3
)
diff --git a/go.sum b/go.sum
index 4007af83a..f063e6d05 100644
--- a/go.sum
+++ b/go.sum
@@ -55,8 +55,8 @@ github.com/Nvveen/Gotty
v0.0.0-20120604004816-cd527374f1e5/go.mod h1:lmUJ/7eu/Q8
github.com/RoaringBitmap/roaring v0.9.4/go.mod
h1:icnadbWcNyfEHlYdr+tDlOTih1Bf/h+rzPpv4sbomAA=
github.com/RoaringBitmap/roaring v1.9.4
h1:yhEIoH4YezLYT04s1nHehNO64EKFTop/wBhxv2QzDdQ=
github.com/RoaringBitmap/roaring v1.9.4/go.mod
h1:6AXUsoIEzDTFFQCe1RbGA6uFONMhvejWj5rqITANK90=
-github.com/SkyAPM/bluge v0.0.0-20250804100126-cccf29a55f01
h1:f56Ybipl9Yh9wFCZc3EoB5OmbW2UJfIpe6e/gsE/DuA=
-github.com/SkyAPM/bluge v0.0.0-20250804100126-cccf29a55f01/go.mod
h1:VfQRtJQEwpooobYuSBadb8QYjTUICH+CX42IJYrfSiM=
+github.com/SkyAPM/bluge v0.0.0-20260625022800-42385daf66b8
h1:kNqj2T61hvMFOblOmkADOf4UWpJPh55jQ4WX2U31zaw=
+github.com/SkyAPM/bluge v0.0.0-20260625022800-42385daf66b8/go.mod
h1:VfQRtJQEwpooobYuSBadb8QYjTUICH+CX42IJYrfSiM=
github.com/SkyAPM/ice v0.0.0-20250619023539-b5173603b0b3
h1:bZvLPihpC1q1AJkmR6ere/aJOr3ev+JCytvjKuL+gE8=
github.com/SkyAPM/ice v0.0.0-20250619023539-b5173603b0b3/go.mod
h1:DoQeb0Ee86LyruZSL77Ddscfk/THJ38x453CRCnGEPI=
github.com/SkyAPM/ktm-ebpf v0.0.0-20260228024820-81a19d950bff
h1:pjcdkEAe+NL2bA0J/mu/Vs+eG+w+KN/BX9yGkOC1f60=
