mirror of
https://github.com/seaweedfs/seaweedfs.git
synced 2026-09-08 15:41:15 +02:00
s3: share one retry allowance across a batch delete (#11001)
Every key in a multi-object delete drives its own retryFilerOp, so a filer that is briefly unhealthy multiplied one op's ~3.1s of backoff by a key count the client picks. The batch now carries a single allowance in its context, sized to one op's worst case; once it is spent the remaining keys fail fast with a per-key error instead of holding the request goroutine. A single-object delete carries no allowance and keeps its full retries. Claude-Session: https://claude.ai/code/session_01BjDWtZsCoZY6x4pdDmGWxU
This commit is contained in:
@@ -447,6 +447,10 @@ func (s3a *S3ApiServer) DeleteMultipleObjectsHandler(w http.ResponseWriter, r *h
|
|||||||
identity, _ = id.(*Identity)
|
identity, _ = id.(*Identity)
|
||||||
}
|
}
|
||||||
|
|
||||||
|
// The keys below each drive their own bounded filer retries, and the client
|
||||||
|
// picks how many keys there are, so the whole batch shares one allowance.
|
||||||
|
r = r.WithContext(withFilerRetryBudget(r.Context(), filerRetryRequestBudget))
|
||||||
|
|
||||||
err = s3a.WithFilerClient(false, func(client filer_pb.SeaweedFilerClient) error {
|
err = s3a.WithFilerClient(false, func(client filer_pb.SeaweedFilerClient) error {
|
||||||
// delete file entries
|
// delete file entries
|
||||||
for _, object := range deleteObjects.Objects {
|
for _, object := range deleteObjects.Objects {
|
||||||
|
|||||||
@@ -15,6 +15,7 @@ import (
|
|||||||
"sort"
|
"sort"
|
||||||
"strconv"
|
"strconv"
|
||||||
"strings"
|
"strings"
|
||||||
|
"sync"
|
||||||
"time"
|
"time"
|
||||||
|
|
||||||
"github.com/seaweedfs/seaweedfs/weed/filer"
|
"github.com/seaweedfs/seaweedfs/weed/filer"
|
||||||
@@ -1375,14 +1376,64 @@ func (s3a *S3ApiServer) repointLatestBeforeDeletion(ctx context.Context, bucket,
|
|||||||
|
|
||||||
// retryAttempts and retryStep tune the bounded retries used when the
|
// retryAttempts and retryStep tune the bounded retries used when the
|
||||||
// load-bearing filer ops in updateLatestVersionAfterDeletion fail with
|
// load-bearing filer ops in updateLatestVersionAfterDeletion fail with
|
||||||
// transient errors. Doubled per attempt, capped at retryCap. Total
|
// transient errors. Doubled per attempt, capped at retryCap, for a
|
||||||
// worst-case wall time ≈ 6.3s before propagating.
|
// worst case of ~3.1s of backoff per op before propagating.
|
||||||
const (
|
const (
|
||||||
updateLatestRetryAttempts = 6
|
updateLatestRetryAttempts = 6
|
||||||
updateLatestRetryStep = 100 * time.Millisecond
|
updateLatestRetryStep = 100 * time.Millisecond
|
||||||
updateLatestRetryCap = 2 * time.Second
|
updateLatestRetryCap = 2 * time.Second
|
||||||
)
|
)
|
||||||
|
|
||||||
|
func retryFilerBackoff(attempt int) time.Duration {
|
||||||
|
backoff := updateLatestRetryStep << (attempt - 1)
|
||||||
|
if backoff <= 0 || backoff > updateLatestRetryCap {
|
||||||
|
return updateLatestRetryCap
|
||||||
|
}
|
||||||
|
return backoff
|
||||||
|
}
|
||||||
|
|
||||||
|
// filerRetryRequestBudget is the total backoff one request may spend across
|
||||||
|
// every retryFilerOp it drives: the worst case of a single op, so a batch
|
||||||
|
// whose length the client chooses waits about as long as one key would.
|
||||||
|
var filerRetryRequestBudget = func() (total time.Duration) {
|
||||||
|
for attempt := 1; attempt < updateLatestRetryAttempts; attempt++ {
|
||||||
|
total += retryFilerBackoff(attempt)
|
||||||
|
}
|
||||||
|
return total
|
||||||
|
}()
|
||||||
|
|
||||||
|
type filerRetryBudgetKey struct{}
|
||||||
|
|
||||||
|
type filerRetryBudget struct {
|
||||||
|
mu sync.Mutex
|
||||||
|
remaining time.Duration
|
||||||
|
}
|
||||||
|
|
||||||
|
// withFilerRetryBudget hands every retryFilerOp reached through ctx one shared
|
||||||
|
// allowance, so per-key backoff no longer multiplies by the number of keys.
|
||||||
|
func withFilerRetryBudget(ctx context.Context, total time.Duration) context.Context {
|
||||||
|
return context.WithValue(ctx, filerRetryBudgetKey{}, &filerRetryBudget{remaining: total})
|
||||||
|
}
|
||||||
|
|
||||||
|
func filerRetryBudgetFrom(ctx context.Context) *filerRetryBudget {
|
||||||
|
budget, _ := ctx.Value(filerRetryBudgetKey{}).(*filerRetryBudget)
|
||||||
|
return budget
|
||||||
|
}
|
||||||
|
|
||||||
|
// take reserves up to d of what is left, reporting false once nothing is.
|
||||||
|
func (b *filerRetryBudget) take(d time.Duration) (time.Duration, bool) {
|
||||||
|
b.mu.Lock()
|
||||||
|
defer b.mu.Unlock()
|
||||||
|
if b.remaining <= 0 {
|
||||||
|
return 0, false
|
||||||
|
}
|
||||||
|
if d > b.remaining {
|
||||||
|
d = b.remaining
|
||||||
|
}
|
||||||
|
b.remaining -= d
|
||||||
|
return d, true
|
||||||
|
}
|
||||||
|
|
||||||
// isRetryableFilerErr reports whether err is worth retrying through
|
// isRetryableFilerErr reports whether err is worth retrying through
|
||||||
// retryFilerOp. Terminal conditions return false so the caller surfaces
|
// retryFilerOp. Terminal conditions return false so the caller surfaces
|
||||||
// them immediately without the backoff delay or the retry-budget
|
// them immediately without the backoff delay or the retry-budget
|
||||||
@@ -1412,7 +1463,7 @@ func isRetryableFilerErr(err error) bool {
|
|||||||
|
|
||||||
func retryFilerOp(ctx context.Context, name string, fn func() error) error {
|
func retryFilerOp(ctx context.Context, name string, fn func() error) error {
|
||||||
var lastErr error
|
var lastErr error
|
||||||
backoff := updateLatestRetryStep
|
budget := filerRetryBudgetFrom(ctx)
|
||||||
for attempt := 1; attempt <= updateLatestRetryAttempts; attempt++ {
|
for attempt := 1; attempt <= updateLatestRetryAttempts; attempt++ {
|
||||||
err := fn()
|
err := fn()
|
||||||
if err == nil {
|
if err == nil {
|
||||||
@@ -1431,9 +1482,17 @@ func retryFilerOp(ctx context.Context, name string, fn func() error) error {
|
|||||||
if attempt == updateLatestRetryAttempts {
|
if attempt == updateLatestRetryAttempts {
|
||||||
break
|
break
|
||||||
}
|
}
|
||||||
|
backoff := retryFilerBackoff(attempt)
|
||||||
|
if budget != nil {
|
||||||
|
granted, ok := budget.take(backoff)
|
||||||
|
if !ok {
|
||||||
|
return fmt.Errorf("%s stopped after %d attempts, request retry allowance spent: %w", name, attempt, lastErr)
|
||||||
|
}
|
||||||
|
backoff = granted
|
||||||
|
}
|
||||||
// Context-aware backoff so a server shutdown / client
|
// Context-aware backoff so a server shutdown / client
|
||||||
// disconnect cancels the worst-case ~6.3s retry budget
|
// disconnect cancels the pending retries immediately
|
||||||
// immediately instead of blocking the goroutine.
|
// instead of blocking the goroutine.
|
||||||
timer := time.NewTimer(backoff)
|
timer := time.NewTimer(backoff)
|
||||||
select {
|
select {
|
||||||
case <-ctx.Done():
|
case <-ctx.Done():
|
||||||
@@ -1441,10 +1500,6 @@ func retryFilerOp(ctx context.Context, name string, fn func() error) error {
|
|||||||
return ctx.Err()
|
return ctx.Err()
|
||||||
case <-timer.C:
|
case <-timer.C:
|
||||||
}
|
}
|
||||||
backoff *= 2
|
|
||||||
if backoff > updateLatestRetryCap {
|
|
||||||
backoff = updateLatestRetryCap
|
|
||||||
}
|
|
||||||
}
|
}
|
||||||
return fmt.Errorf("%s exhausted %d retries: %w", name, updateLatestRetryAttempts, lastErr)
|
return fmt.Errorf("%s exhausted %d retries: %w", name, updateLatestRetryAttempts, lastErr)
|
||||||
}
|
}
|
||||||
|
|||||||
@@ -168,3 +168,78 @@ func TestRetryFilerOp_TerminalErrorsShortCircuit(t *testing.T) {
|
|||||||
})
|
})
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
|
// TestFilerRetryBudget_ClampsToWhatIsLeft covers the allowance arithmetic:
|
||||||
|
// a reservation larger than the remainder is trimmed, and once nothing is
|
||||||
|
// left the caller is told to stop rather than handed a zero wait.
|
||||||
|
func TestFilerRetryBudget_ClampsToWhatIsLeft(t *testing.T) {
|
||||||
|
b := &filerRetryBudget{remaining: 150 * time.Millisecond}
|
||||||
|
|
||||||
|
granted, ok := b.take(100 * time.Millisecond)
|
||||||
|
require.True(t, ok)
|
||||||
|
assert.Equal(t, 100*time.Millisecond, granted)
|
||||||
|
|
||||||
|
granted, ok = b.take(200 * time.Millisecond)
|
||||||
|
require.True(t, ok)
|
||||||
|
assert.Equal(t, 50*time.Millisecond, granted, "trimmed to the remainder")
|
||||||
|
|
||||||
|
_, ok = b.take(time.Millisecond)
|
||||||
|
assert.False(t, ok, "allowance spent")
|
||||||
|
}
|
||||||
|
|
||||||
|
// TestFilerRetryRequestBudget_MatchesOneOpWorstCase pins the request-wide
|
||||||
|
// allowance to what a single retryFilerOp can sleep for, so a batch delete
|
||||||
|
// adds the wait of one key rather than one per key.
|
||||||
|
func TestFilerRetryRequestBudget_MatchesOneOpWorstCase(t *testing.T) {
|
||||||
|
var singleOp time.Duration
|
||||||
|
for attempt := 1; attempt < updateLatestRetryAttempts; attempt++ {
|
||||||
|
singleOp += retryFilerBackoff(attempt)
|
||||||
|
}
|
||||||
|
assert.Equal(t, singleOp, filerRetryRequestBudget)
|
||||||
|
assert.Equal(t, 3100*time.Millisecond, filerRetryRequestBudget)
|
||||||
|
}
|
||||||
|
|
||||||
|
// TestRetryFilerOp_SharedBudgetAcrossBatch is the batch-delete shape: many
|
||||||
|
// keys, each driving its own retryFilerOp against a filer that always fails
|
||||||
|
// retryably. Without a shared allowance every key pays the full per-op
|
||||||
|
// backoff, so the wait scales with a key count the client picks. The
|
||||||
|
// allowance here is deliberately small so the test never sleeps for the
|
||||||
|
// production budget.
|
||||||
|
func TestRetryFilerOp_SharedBudgetAcrossBatch(t *testing.T) {
|
||||||
|
const keys = 200
|
||||||
|
const allowance = 250 * time.Millisecond
|
||||||
|
|
||||||
|
ctx := withFilerRetryBudget(context.Background(), allowance)
|
||||||
|
calls := 0
|
||||||
|
start := time.Now()
|
||||||
|
for i := 0; i < keys; i++ {
|
||||||
|
err := retryFilerOp(ctx, "test", func() error {
|
||||||
|
calls++
|
||||||
|
return errors.New("transient")
|
||||||
|
})
|
||||||
|
require.Error(t, err)
|
||||||
|
}
|
||||||
|
elapsed := time.Since(start)
|
||||||
|
|
||||||
|
assert.LessOrEqual(t, calls, keys+updateLatestRetryAttempts, "only the keys that fit the allowance retry")
|
||||||
|
assert.Less(t, elapsed, 2*allowance, "the whole batch stays inside one allowance")
|
||||||
|
}
|
||||||
|
|
||||||
|
// TestRetryFilerOp_SpentBudgetStopsAfterFirstAttempt confirms a key that
|
||||||
|
// arrives after the allowance is gone fails immediately, reporting why, and
|
||||||
|
// still gets its one real attempt at the filer.
|
||||||
|
func TestRetryFilerOp_SpentBudgetStopsAfterFirstAttempt(t *testing.T) {
|
||||||
|
ctx := withFilerRetryBudget(context.Background(), 0)
|
||||||
|
calls := 0
|
||||||
|
start := time.Now()
|
||||||
|
err := retryFilerOp(ctx, "test", func() error {
|
||||||
|
calls++
|
||||||
|
return errors.New("transient")
|
||||||
|
})
|
||||||
|
|
||||||
|
require.Error(t, err)
|
||||||
|
assert.Equal(t, 1, calls, "the op still runs once")
|
||||||
|
assert.Less(t, time.Since(start), 50*time.Millisecond, "no backoff once the allowance is spent")
|
||||||
|
assert.Contains(t, err.Error(), "retry allowance spent")
|
||||||
|
assert.Contains(t, err.Error(), "transient", "underlying error preserved")
|
||||||
|
}
|
||||||
|
|||||||
Reference in New Issue
Block a user