diff --git a/drivers/baidu_netdisk/driver.go b/drivers/baidu_netdisk/driver.go
index fe77aca38b..cda878605b 100644
--- a/drivers/baidu_netdisk/driver.go
+++ b/drivers/baidu_netdisk/driver.go
@@ -33,7 +33,8 @@ type BaiduNetdisk struct {
Addition
uploadThread int
- vipType int // 会员类型，0普通用户(4G/4M)、1普通会员(10G/16M)、2超级会员(20G/32M)
+ vipType int // 会员类型，0普通用户(4G/4M)、1普通会员(10G/16M)、2超级会员(20G/32M)
+ cooldown *cooldown // shared pause after Baidu rate limiting (31034)
}
var ErrUploadIDExpired = errors.New("uploadid expired")
@@ -47,6 +48,9 @@ func (d *BaiduNetdisk) GetAddition() driver.Additional {
}
func (d *BaiduNetdisk) Init(ctx context.Context) error {
+ if d.cooldown == nil {
+ d.cooldown = newCooldown()
+ }
d.uploadThread, _ = strconv.Atoi(d.UploadThread)
if d.uploadThread < 1 {
d.uploadThread, d.UploadThread = 1, "1"
@@ -110,7 +114,7 @@ func (d *BaiduNetdisk) Move(ctx context.Context, srcObj, dstDir model.Obj) (mode
"newname": srcObj.GetName(),
},
}
- _, err := d.manage("move", data)
+ _, err := d.manage(ctx, "move", data)
if err != nil {
return nil, err
}
@@ -129,7 +133,7 @@ func (d *BaiduNetdisk) Rename(ctx context.Context, srcObj model.Obj, newName str
"newname": newName,
},
}
- _, err := d.manage("rename", data)
+ _, err := d.manage(ctx, "rename", data)
if err != nil {
return nil, err
}
@@ -151,13 +155,18 @@ func (d *BaiduNetdisk) Copy(ctx context.Context, srcObj, dstDir model.Obj) error
"newname": srcObj.GetName(),
},
}
- _, err := d.manage("copy", data)
+ _, err := d.manage(ctx, "copy", data)
return err
}
func (d *BaiduNetdisk) Remove(ctx context.Context, obj model.Obj) error {
data := []string{obj.GetPath()}
- _, err := d.manage("delete", data)
+ _, err := d.manage(ctx, "delete", data)
+ if e, ok := asAPIError(err); ok && e.NotExist() {
+ // already gone: deleting is idempotent
+ log.Debugf("[baidu_netdisk] %s already removed (errno %d)", obj.GetPath(), e.Errno)
+ return nil
+ }
return err
}
diff --git a/drivers/baidu_netdisk/ratelimit.go b/drivers/baidu_netdisk/ratelimit.go
new file mode 100644
index 0000000000..98335a3d3b
--- /dev/null
+++ b/drivers/baidu_netdisk/ratelimit.go
@@ -0,0 +1,287 @@
+package baidu_netdisk
+
+import (
+ "context"
+ "errors"
+ "fmt"
+ "math/rand"
+ "sync"
+ "time"
+
+ "github.com/OpenListTeam/OpenList/v4/internal/errs"
+ "github.com/OpenListTeam/OpenList/v4/pkg/utils"
+ log "github.com/sirupsen/logrus"
+)
+
+// Baidu error codes handled specially.
+const (
+ errnoBatchFailed = 12 // filemanager: some items of the batch failed, see info[].errno
+ errnoNotExist = -9 // file or directory does not exist
+ errnoPCSNotExist = 31066 // PCS: file does not exist
+ errnoRateLimited = 31034 // hit API frequency control
+ cooldownBase = 8 * time.Second
+ cooldownMax = 30 * time.Second
+ // maxCooldownWaitPerCall bounds how long a single API call may wait for
+ // the breaker. Past that the call fails with a rate-limit error so callers
+ // (e.g. S3 clients receiving SlowDown) can back off on their own.
+ maxCooldownWaitPerCall = 35 * time.Second
+)
+
+// apiItem is one entry of a filemanager response's info list.
+type apiItem struct {
+ Errno int `json:"errno"`
+ Path string `json:"path"`
+}
+
+// apiError is a non-zero errno returned by the Baidu API.
+type apiError struct {
+ URL string
+ Errno int
+ Items []apiItem // failed items of a filemanager batch, if any
+}
+
+func (e *apiError) Error() string {
+ if len(e.Items) > 0 {
+ return fmt.Sprintf("req: [%s] ,errno: %d, item errno: %d (path %s), refer to https://pan.baidu.com/union/doc/",
+ e.URL, e.Errno, e.Items[0].Errno, e.Items[0].Path)
+ }
+ return fmt.Sprintf("req: [%s] ,errno: %d, refer to https://pan.baidu.com/union/doc/", e.URL, e.Errno)
+}
+
+// RateLimited reports whether Baidu rejected the request with frequency control,
+// either directly or for any item of a batch.
+func (e *apiError) RateLimited() bool {
+ if e.Errno == errnoRateLimited {
+ return true
+ }
+ for _, it := range e.Items {
+ if it.Errno == errnoRateLimited {
+ return true
+ }
+ }
+ return false
+}
+
+// NotExist reports whether the request failed only because the target(s) do not exist.
+func (e *apiError) NotExist() bool {
+ if isNotExistErrno(e.Errno) {
+ return true
+ }
+ if e.Errno != errnoBatchFailed || len(e.Items) == 0 {
+ return false
+ }
+ for _, it := range e.Items {
+ if !isNotExistErrno(it.Errno) {
+ return false
+ }
+ }
+ return true
+}
+
+// Unwrap lets errors.Is(err, errs.TemporaryCapacity) match rate-limit errors,
+// which the S3 server maps to 503 SlowDown.
+func (e *apiError) Unwrap() error {
+ if e.RateLimited() {
+ return errs.TemporaryCapacity
+ }
+ return nil
+}
+
+func isNotExistErrno(n int) bool { return n == errnoNotExist || n == errnoPCSNotExist }
+
+// parseAPIError returns nil when body has errno 0. For a failed filemanager
+// batch it keeps the failed info[] items, which carry the real cause.
+func parseAPIError(furl string, body []byte) *apiError {
+ var r struct {
+ Errno int `json:"errno"`
+ Info []apiItem `json:"info"`
+ }
+ if err := utils.Json.Unmarshal(body, &r); err != nil {
+ // keep the previous behaviour for odd bodies: only look at errno
+ r.Errno = utils.Json.Get(body, "errno").ToInt()
+ r.Info = nil
+ }
+ if r.Errno == 0 {
+ return nil
+ }
+ e := &apiError{URL: furl, Errno: r.Errno}
+ for _, it := range r.Info {
+ if it.Errno != 0 {
+ e.Items = append(e.Items, it)
+ }
+ }
+ return e
+}
+
+func asAPIError(err error) (*apiError, bool) {
+ var e *apiError
+ ok := errors.As(err, &e)
+ return e, ok
+}
+
+// outcome of a request made under a permit.
+type outcome int
+
+const (
+ outcomeSuccess outcome = iota
+ outcomeRateLimited
+ outcomeOther // any other error, including ctx cancellation
+)
+
+// cooldown is a small circuit breaker shared by all requests of one driver
+// instance. When Baidu answers 31034 the breaker opens and every request
+// pauses. When the pause ends exactly one request is let through as a probe
+// while the rest keep waiting:
+// - probe succeeds: the breaker closes and all waiters proceed;
+// - probe gets 31034: the pause is extended (doubling, capped) and the next
+// expiry probes again;
+// - probe fails otherwise or is cancelled: the probe slot passes to the next
+// waiter.
+//
+// There is no limit while the breaker is closed.
+type cooldown struct {
+ mu sync.Mutex
+ tripped bool // breaker open
+ until time.Time // end of the current pause
+ penalty time.Duration // length of the next pause
+ probing bool // a probe is in flight
+ changed chan struct{} // closed and replaced on every state change
+ probes int // probes granted (for tests and logs)
+
+ base, max time.Duration // pause bounds
+ maxWait time.Duration // per-call wait budget
+
+ now func() time.Time
+ jitter func() time.Duration
+}
+
+func newCooldown() *cooldown {
+ c := &cooldown{
+ penalty: cooldownBase,
+ base: cooldownBase,
+ max: cooldownMax,
+ maxWait: maxCooldownWaitPerCall,
+ changed: make(chan struct{}),
+ now: time.Now,
+ }
+ // up to 1/8 of the base pause (1s by default), so instances do not resume in lockstep
+ c.jitter = func() time.Duration { return time.Duration(rand.Int63n(int64(c.base/8) + 1)) }
+ return c
+}
+
+// budget is the per-call wait limit.
+func (c *cooldown) budget() time.Duration {
+ if c == nil {
+ return 0
+ }
+ return c.maxWait
+}
+
+// permit is held by one request between acquire and done.
+type permit struct {
+ c *cooldown
+ probe bool
+ once sync.Once
+}
+
+// done reports the request's outcome. It must be called exactly once per
+// permit; extra calls are ignored. A nil permit is a no-op.
+func (p *permit) done(o outcome) {
+ if p == nil || p.c == nil {
+ return
+ }
+ p.once.Do(func() { p.c.report(p.probe, o) })
+}
+
+// acquire waits until the request may be sent. It returns ok=false without
+// sending once waiting would exceed budget, and ctx's error if ctx ends first.
+// waited is the time spent waiting either way.
+func (c *cooldown) acquire(ctx context.Context, budget time.Duration) (p *permit, waited time.Duration, ok bool, err error) {
+ if c == nil {
+ return nil, 0, true, nil
+ }
+ start := c.now()
+ for {
+ c.mu.Lock()
+ now := c.now()
+ waited = now.Sub(start)
+ if !c.tripped {
+ c.mu.Unlock()
+ return &permit{c: c}, waited, true, nil
+ }
+ wait := c.until.Sub(now) // > 0: pause still running
+ if wait <= 0 && !c.probing {
+ c.probing = true
+ c.probes++
+ c.mu.Unlock()
+ return &permit{c: c, probe: true}, waited, true, nil
+ }
+ ch := c.changed
+ c.mu.Unlock()
+
+ left := budget - waited
+ if left <= 0 || wait > left {
+ return nil, waited, false, nil
+ }
+ if wait <= 0 { // a probe is in flight: wait for its report
+ wait = left
+ }
+ t := time.NewTimer(wait)
+ select {
+ case <-ch:
+ case <-t.C:
+ case <-ctx.Done():
+ t.Stop()
+ return nil, c.now().Sub(start), false, ctx.Err()
+ }
+ t.Stop()
+ }
+}
+
+func (c *cooldown) report(probe bool, o outcome) {
+ c.mu.Lock()
+ defer c.mu.Unlock()
+ now := c.now()
+ switch {
+ case probe:
+ c.probing = false
+ switch o {
+ case outcomeSuccess:
+ c.tripped = false
+ c.penalty = c.base
+ log.Infof("[baidu_netdisk] rate limit probe succeeded, resuming requests")
+ case outcomeRateLimited:
+ d := c.pause(now)
+ log.Warnf("[baidu_netdisk] rate limit probe still throttled (31034), pausing requests for %s", d.Round(time.Millisecond))
+ default:
+ log.Debugf("[baidu_netdisk] rate limit probe failed without a verdict, handing over")
+ }
+ c.broadcast()
+ case o == outcomeRateLimited && !c.tripped:
+ c.tripped = true
+ d := c.pause(now)
+ log.Warnf("[baidu_netdisk] hit rate limit (31034), pausing requests for %s", d.Round(time.Millisecond))
+ c.broadcast()
+ }
+ // Other outcomes of non-probe requests, e.g. 31034 for requests sent
+ // before the breaker opened, change nothing.
+}
+
+// pause starts a new pause and escalates the next one. c.mu must be held.
+func (c *cooldown) pause(now time.Time) time.Duration {
+ if c.penalty <= 0 {
+ c.penalty = c.base
+ }
+ d := c.penalty + c.jitter()
+ c.until = now.Add(d)
+ c.penalty *= 2
+ if c.penalty > c.max {
+ c.penalty = c.max
+ }
+ return d
+}
+
+func (c *cooldown) broadcast() {
+ close(c.changed)
+ c.changed = make(chan struct{})
+}
diff --git a/drivers/baidu_netdisk/ratelimit_test.go b/drivers/baidu_netdisk/ratelimit_test.go
new file mode 100644
index 0000000000..13b3ea897c
--- /dev/null
+++ b/drivers/baidu_netdisk/ratelimit_test.go
@@ -0,0 +1,445 @@
+package baidu_netdisk
+
+import (
+ "context"
+ "errors"
+ "io"
+ "net/http"
+ "runtime"
+ "strings"
+ "sync"
+ "sync/atomic"
+ "testing"
+ "time"
+
+ "github.com/OpenListTeam/OpenList/v4/drivers/base"
+ "github.com/OpenListTeam/OpenList/v4/internal/errs"
+ "github.com/OpenListTeam/OpenList/v4/internal/model"
+ "github.com/go-resty/resty/v2"
+)
+
+// Real body captured from Baidu while deleting quickly (path shortened).
+const rateLimitedBatchBody = `{"errno":12,"info":[{"errno":31034,"path":"\/alist\/x\/q0120"}],"request_id":9111630116005502295}`
+
+func TestParseAPIError(t *testing.T) {
+ cases := []struct {
+ name string
+ body string
+ wantNil bool
+ wantRL, wantNotExt bool
+ }{
+ {name: "success", body: `{"errno":0,"info":[{"errno":0,"path":"/a"}]}`, wantNil: true},
+ {name: "no errno field", body: `{"list":[]}`, wantNil: true},
+ {name: "batch rate limited", body: rateLimitedBatchBody, wantRL: true},
+ {name: "outer rate limited", body: `{"errno":31034}`, wantRL: true},
+ {name: "batch not exist", body: `{"errno":12,"info":[{"errno":-9,"path":"/a"}]}`, wantNotExt: true},
+ {name: "batch pcs not exist", body: `{"errno":12,"info":[{"errno":31066,"path":"/a"}]}`, wantNotExt: true},
+ {name: "outer not exist", body: `{"errno":-9}`, wantNotExt: true},
+ {name: "batch mixed", body: `{"errno":12,"info":[{"errno":-9,"path":"/a"},{"errno":-7,"path":"/b"}]}`},
+ {name: "batch without info", body: `{"errno":12}`},
+ {name: "token", body: `{"errno":-6}`},
+ {name: "info not a list", body: `{"errno":2,"info":"x"}`},
+ }
+ for _, c := range cases {
+ t.Run(c.name, func(t *testing.T) {
+ e := parseAPIError("u", []byte(c.body))
+ if c.wantNil {
+ if e != nil {
+ t.Fatalf("want nil, got %v", e)
+ }
+ return
+ }
+ if e == nil {
+ t.Fatal("want error, got nil")
+ }
+ if e.RateLimited() != c.wantRL {
+ t.Errorf("RateLimited = %v, want %v", e.RateLimited(), c.wantRL)
+ }
+ if e.NotExist() != c.wantNotExt {
+ t.Errorf("NotExist = %v, want %v", e.NotExist(), c.wantNotExt)
+ }
+ if got := errors.Is(e, errs.TemporaryCapacity); got != c.wantRL {
+ t.Errorf("errors.Is(TemporaryCapacity) = %v, want %v", got, c.wantRL)
+ }
+ })
+ }
+ e := parseAPIError("https://pan.baidu.com/rest/2.0/xpan/file", []byte(rateLimitedBatchBody))
+ if !strings.Contains(e.Error(), "31034") || !strings.Contains(e.Error(), "q0120") {
+ t.Errorf("message should carry item errno and path: %s", e.Error())
+ }
+}
+
+func newTestBreaker(base, max, maxWait time.Duration) *cooldown {
+ c := newCooldown()
+ c.base, c.penalty, c.max, c.maxWait = base, base, max, maxWait
+ c.jitter = func() time.Duration { return 0 }
+ return c
+}
+
+func trip(c *cooldown) { c.report(false, outcomeRateLimited) }
+
+func checkNoLeak(t *testing.T, before int) {
+ t.Helper()
+ deadline := time.Now().Add(2 * time.Second)
+ for runtime.NumGoroutine() > before && time.Now().Before(deadline) {
+ time.Sleep(10 * time.Millisecond)
+ }
+ if n := runtime.NumGoroutine(); n > before {
+ t.Fatalf("goroutines: %d before, %d after", before, n)
+ }
+}
+
+// probeScenario starts n waiters on a tripped breaker. Each probe reports the
+// next verdict from verdicts (the last one repeats); non-probe holders report
+// success. It records grants, non-probe requests let through before the
+// successful probe reported (early) and overlapping probes (maxInflight).
+type probeScenario struct {
+ mu sync.Mutex
+ probeGrants []time.Time
+ probeDone []time.Time
+ released time.Time
+ early int
+ inflight int
+ maxInflight int
+ granted int
+}
+
+func runProbeScenario(t *testing.T, c *cooldown, n int, hold time.Duration, verdicts ...outcome) *probeScenario {
+ t.Helper()
+ sc := &probeScenario{}
+ var wg sync.WaitGroup
+ for i := 0; i < n; i++ {
+ wg.Add(1)
+ go func() {
+ defer wg.Done()
+ p, _, ok, err := c.acquire(context.Background(), 10*time.Second)
+ if !ok || err != nil {
+ t.Errorf("acquire: ok=%v err=%v", ok, err)
+ return
+ }
+ sc.mu.Lock()
+ sc.granted++
+ if p.probe {
+ sc.probeGrants = append(sc.probeGrants, time.Now())
+ sc.inflight++
+ if sc.inflight > sc.maxInflight {
+ sc.maxInflight = sc.inflight
+ }
+ v := verdicts[min(len(sc.probeGrants)-1, len(verdicts)-1)]
+ sc.mu.Unlock()
+ time.Sleep(hold)
+ sc.mu.Lock()
+ sc.inflight--
+ sc.probeDone = append(sc.probeDone, time.Now())
+ if v == outcomeSuccess {
+ sc.released = time.Now()
+ }
+ sc.mu.Unlock()
+ p.done(v)
+ return
+ }
+ if sc.released.IsZero() {
+ sc.early++
+ }
+ sc.mu.Unlock()
+ p.done(outcomeSuccess)
+ }()
+ }
+ wg.Wait()
+ return sc
+}
+
+func TestBreakerLetsOneProbeThroughThenReleasesAll(t *testing.T) {
+ before := runtime.NumGoroutine()
+ c := newTestBreaker(100*time.Millisecond, time.Second, 10*time.Second)
+ start := time.Now()
+ trip(c)
+ sc := runProbeScenario(t, c, 32, 50*time.Millisecond, outcomeSuccess)
+ if len(sc.probeGrants) != 1 || sc.maxInflight != 1 {
+ t.Fatalf("probes=%d maxInflight=%d, want exactly 1", len(sc.probeGrants), sc.maxInflight)
+ }
+ if sc.early != 0 {
+ t.Fatalf("%d requests went out before the probe succeeded", sc.early)
+ }
+ if sc.granted != 32 {
+ t.Fatalf("granted %d of 32", sc.granted)
+ }
+ if d := sc.probeGrants[0].Sub(start); d < 100*time.Millisecond {
+ t.Fatalf("probe after %s, before the pause ended", d)
+ }
+ if c.tripped || c.penalty != c.base {
+ t.Fatalf("breaker should be closed and reset: tripped=%v penalty=%s", c.tripped, c.penalty)
+ }
+ checkNoLeak(t, before)
+}
+
+func TestBreakerThrottledProbeRearmsPause(t *testing.T) {
+ before := runtime.NumGoroutine()
+ c := newTestBreaker(60*time.Millisecond, time.Second, 10*time.Second)
+ trip(c) // pause 60ms, next 120ms
+ sc := runProbeScenario(t, c, 32, 20*time.Millisecond, outcomeRateLimited, outcomeRateLimited, outcomeSuccess)
+ if len(sc.probeGrants) != 3 || sc.maxInflight != 1 || sc.early != 0 || sc.granted != 32 {
+ t.Fatalf("probes=%d maxInflight=%d early=%d granted=%d", len(sc.probeGrants), sc.maxInflight, sc.early, sc.granted)
+ }
+ for i, want := range []time.Duration{120 * time.Millisecond, 240 * time.Millisecond} {
+ if gap := sc.probeGrants[i+1].Sub(sc.probeDone[i]); gap < want {
+ t.Fatalf("probe %d came %s after the throttled probe, want >= %s", i+2, gap, want)
+ }
+ }
+ checkNoLeak(t, before)
+}
+
+func TestBreakerHandsProbeOverOnFailure(t *testing.T) {
+ before := runtime.NumGoroutine()
+ c := newTestBreaker(50*time.Millisecond, time.Second, 10*time.Second)
+ trip(c)
+ // the first probe fails without a verdict (ctx cancelled / network error)
+ sc := runProbeScenario(t, c, 32, 10*time.Millisecond, outcomeOther, outcomeSuccess)
+ if len(sc.probeGrants) != 2 || sc.maxInflight != 1 || sc.early != 0 || sc.granted != 32 {
+ t.Fatalf("probes=%d maxInflight=%d early=%d granted=%d", len(sc.probeGrants), sc.maxInflight, sc.early, sc.granted)
+ }
+ if gap := sc.probeGrants[1].Sub(sc.probeDone[0]); gap > 30*time.Millisecond {
+ t.Fatalf("probe handed over after %s, want immediately", gap)
+ }
+ checkNoLeak(t, before)
+}
+
+func TestBreakerCancelledProbeHolderHandsOver(t *testing.T) {
+ c := newTestBreaker(20*time.Millisecond, time.Second, 10*time.Second)
+ trip(c)
+ time.Sleep(30 * time.Millisecond)
+ p, _, ok, _ := c.acquire(context.Background(), time.Second)
+ if !ok || !p.probe {
+ t.Fatal("expected to become the probe")
+ }
+ got := make(chan *permit)
+ go func() {
+ q, _, _, _ := c.acquire(context.Background(), time.Second)
+ got <- q
+ }()
+ select {
+ case <-got:
+ t.Fatal("second request must wait while a probe is in flight")
+ case <-time.After(50 * time.Millisecond):
+ }
+ p.done(outcomeOther) // e.g. the probe's ctx was cancelled
+ p.done(outcomeSuccess)
+ select {
+ case q := <-got:
+ if q == nil || !q.probe {
+ t.Fatal("waiter should take over the probe")
+ }
+ if !c.tripped {
+ t.Fatal("extra done() calls must be ignored")
+ }
+ q.done(outcomeSuccess)
+ case <-time.After(time.Second):
+ t.Fatal("probe slot was not handed over")
+ }
+}
+
+func TestBreakerBudgetAndContext(t *testing.T) {
+ before := runtime.NumGoroutine()
+ c := newTestBreaker(10*time.Second, 30*time.Second, 35*time.Second)
+ if p, _, ok, err := c.acquire(context.Background(), time.Second); !ok || err != nil || p.probe {
+ t.Fatal("closed breaker must let requests through")
+ } else {
+ p.done(outcomeSuccess)
+ }
+ trip(c)
+ start := time.Now()
+ if _, _, ok, err := c.acquire(context.Background(), time.Second); ok || err != nil {
+ t.Fatalf("pause beyond budget: ok=%v err=%v", ok, err)
+ }
+ if time.Since(start) > 50*time.Millisecond {
+ t.Fatal("pause beyond budget should fail fast")
+ }
+ ctx, cancel := context.WithTimeout(context.Background(), 50*time.Millisecond)
+ defer cancel()
+ if _, _, _, err := c.acquire(ctx, time.Minute); !errors.Is(err, context.DeadlineExceeded) {
+ t.Fatalf("want ctx error, got %v", err)
+ }
+
+ // a probe that never reports: waiters give up at their budget
+ c2 := newTestBreaker(10*time.Millisecond, time.Second, time.Second)
+ trip(c2)
+ time.Sleep(20 * time.Millisecond)
+ p, _, _, _ := c2.acquire(context.Background(), time.Second)
+ start = time.Now()
+ _, w, ok, err := c2.acquire(context.Background(), 100*time.Millisecond)
+ if ok || err != nil || w < 90*time.Millisecond || time.Since(start) > 500*time.Millisecond {
+ t.Fatalf("waiting on a stuck probe: ok=%v err=%v waited=%s", ok, err, w)
+ }
+ p.done(outcomeSuccess)
+
+ var nilC *cooldown
+ if p, _, ok, err := nilC.acquire(context.Background(), 0); !ok || err != nil {
+ t.Fatal("nil breaker must be a no-op")
+ } else {
+ p.done(outcomeRateLimited)
+ }
+ checkNoLeak(t, before)
+}
+
+func TestBreakerEscalation(t *testing.T) {
+ now := time.Unix(1000, 0)
+ c := newTestBreaker(8*time.Second, 30*time.Second, 35*time.Second)
+ c.now = func() time.Time { return now }
+ pauseLen := func() time.Duration { return c.until.Sub(now) }
+
+ trip(c)
+ if pauseLen() != 8*time.Second {
+ t.Fatalf("first pause %s", pauseLen())
+ }
+ // stragglers sent before the trip must not extend or escalate
+ trip(c)
+ c.report(false, outcomeSuccess)
+ if pauseLen() != 8*time.Second || c.penalty != 16*time.Second || !c.tripped {
+ t.Fatalf("straggler changed state: pause=%s penalty=%s", pauseLen(), c.penalty)
+ }
+ for _, want := range []time.Duration{16 * time.Second, 30 * time.Second, 30 * time.Second} {
+ c.report(true, outcomeRateLimited)
+ if pauseLen() != want {
+ t.Fatalf("pause %s, want %s", pauseLen(), want)
+ }
+ }
+ c.report(true, outcomeSuccess)
+ if c.tripped || c.penalty != 8*time.Second {
+ t.Fatalf("probe success should close and reset: tripped=%v penalty=%s", c.tripped, c.penalty)
+ }
+ trip(c)
+ if pauseLen() != 8*time.Second {
+ t.Fatalf("pause after reset %s", pauseLen())
+ }
+}
+
+// fakeBaidu answers filemanager calls: 31034 for every request while the
+// throttle window is open, then success. Deleting an unknown path gives -9.
+type fakeBaidu struct {
+ mu sync.Mutex
+ openUntil time.Time
+ window time.Duration
+ threshold int // requests allowed before the window opens
+ calls atomic.Int64
+ throttled atomic.Int64
+ missing string
+}
+
+func (f *fakeBaidu) RoundTrip(r *http.Request) (*http.Response, error) {
+ f.calls.Add(1)
+ _ = r.ParseForm()
+ body := `{"errno":0,"info":[{"errno":0,"path":"/x"}],"request_id":1}`
+ f.mu.Lock()
+ now := time.Now()
+ if f.missing != "" && strings.Contains(r.PostForm.Get("filelist"), f.missing) {
+ body = `{"errno":12,"info":[{"errno":-9,"path":"/missing"}],"request_id":1}`
+ } else if now.Before(f.openUntil) {
+ body = rateLimitedBatchBody
+ f.throttled.Add(1)
+ } else if f.threshold > 0 {
+ f.threshold--
+ if f.threshold == 0 {
+ f.openUntil = now.Add(f.window)
+ }
+ }
+ f.mu.Unlock()
+ return &http.Response{
+ StatusCode: 200,
+ Header: http.Header{"Content-Type": []string{"application/json"}},
+ Body: io.NopCloser(strings.NewReader(body)),
+ Request: r,
+ }, nil
+}
+
+func withFakeBaidu(t *testing.T, f *fakeBaidu) {
+ t.Helper()
+ old := base.RestyClient
+ base.RestyClient = resty.New().SetTransport(f)
+ t.Cleanup(func() { base.RestyClient = old })
+}
+
+func testDriver(base, max, maxWait time.Duration) *BaiduNetdisk {
+ d := &BaiduNetdisk{cooldown: newCooldown()}
+ d.cooldown.base, d.cooldown.penalty, d.cooldown.max, d.cooldown.maxWait = base, base, max, maxWait
+ return d
+}
+
+func TestRemoveRidesOutRateLimitWindow(t *testing.T) {
+ f := &fakeBaidu{threshold: 20, window: 400 * time.Millisecond}
+ withFakeBaidu(t, f)
+ d := testDriver(150*time.Millisecond, time.Second, 3*time.Second)
+
+ const n, conc = 200, 32
+ sem := make(chan struct{}, conc)
+ var wg sync.WaitGroup
+ var failed atomic.Int64
+ start := time.Now()
+ for i := 0; i < n; i++ {
+ wg.Add(1)
+ sem <- struct{}{}
+ go func() {
+ defer wg.Done()
+ defer func() { <-sem }()
+ obj := &model.Object{Path: "/x"}
+ if err := d.Remove(context.Background(), obj); err != nil {
+ failed.Add(1)
+ }
+ }()
+ }
+ wg.Wait()
+ t.Logf("%d deletes, %d calls to baidu, %d throttled, %d probes, %s", n, f.calls.Load(), f.throttled.Load(), d.cooldown.probes, time.Since(start))
+ if failed.Load() != 0 {
+ t.Fatalf("%d deletes failed", failed.Load())
+ }
+ if f.throttled.Load() == 0 {
+ t.Fatal("test did not exercise the window")
+ }
+ // only requests already in flight when the breaker opened, plus one probe
+ // per pause, may reach Baidu during the window
+ if max := int64(conc + d.cooldown.probes); f.throttled.Load() > max {
+ t.Fatalf("throttled requests = %d, want <= %d (in-flight + probes)", f.throttled.Load(), max)
+ }
+}
+
+func TestRemoveReturnsRateLimitErrorWhenWindowOutlastsBudget(t *testing.T) {
+ f := &fakeBaidu{openUntil: time.Now().Add(time.Hour)}
+ withFakeBaidu(t, f)
+ d := testDriver(100*time.Millisecond, 200*time.Millisecond, 350*time.Millisecond)
+
+ start := time.Now()
+ err := d.Remove(context.Background(), &model.Object{Path: "/x"})
+ if !errors.Is(err, errs.TemporaryCapacity) {
+ t.Fatalf("want rate-limit error, got %v", err)
+ }
+ if el := time.Since(start); el > 2*time.Second {
+ t.Fatalf("call blocked too long: %s", el)
+ }
+ if c := f.calls.Load(); c != 3 {
+ t.Fatalf("calls = %d, want 3 (initial + 2 retries after cooldowns)", c)
+ }
+}
+
+func TestRemoveTreatsMissingAsSuccess(t *testing.T) {
+ f := &fakeBaidu{missing: "gone"}
+ withFakeBaidu(t, f)
+ d := testDriver(time.Second, time.Second, time.Second)
+ if err := d.Remove(context.Background(), &model.Object{Path: "/gone"}); err != nil {
+ t.Fatalf("remove of missing file: %v", err)
+ }
+ if c := f.calls.Load(); c != 1 {
+ t.Fatalf("missing file should not be retried, calls = %d", c)
+ }
+}
+
+func TestRemoveHonorsContextDuringCooldown(t *testing.T) {
+ f := &fakeBaidu{openUntil: time.Now().Add(time.Hour)}
+ withFakeBaidu(t, f)
+ d := testDriver(10*time.Second, 10*time.Second, 30*time.Second)
+ ctx, cancel := context.WithTimeout(context.Background(), 100*time.Millisecond)
+ defer cancel()
+ start := time.Now()
+ err := d.Remove(ctx, &model.Object{Path: "/x"})
+ if err == nil || time.Since(start) > 2*time.Second {
+ t.Fatalf("err=%v after %s, want quick ctx error", err, time.Since(start))
+ }
+}
diff --git a/drivers/baidu_netdisk/util.go b/drivers/baidu_netdisk/util.go
index 31f3125798..5a807f2f14 100644
--- a/drivers/baidu_netdisk/util.go
+++ b/drivers/baidu_netdisk/util.go
@@ -95,9 +95,26 @@ func (d *BaiduNetdisk) _refreshToken() error {
}
func (d *BaiduNetdisk) request(furl string, method string, callback base.ReqCallback, resp interface{}) ([]byte, error) {
- var result []byte
+ return d.requestCtx(context.Background(), furl, method, callback, resp)
+}
+
+func (d *BaiduNetdisk) requestCtx(ctx context.Context, furl string, method string, callback base.ReqCallback, resp interface{}) ([]byte, error) {
+ var body []byte
+ var waited time.Duration
err := retry.Do(func() error {
- req := base.RestyClient.R()
+ // Wait for the shared rate-limit breaker, but never longer than the
+ // per-call budget in total.
+ p, w, ok, err := d.cooldown.acquire(ctx, d.cooldown.budget()-waited)
+ waited += w
+ if err != nil {
+ return retry.Unrecoverable(err)
+ }
+ if !ok {
+ return retry.Unrecoverable(&apiError{URL: furl, Errno: errnoRateLimited})
+ }
+ result := outcomeOther
+ defer func() { p.done(result) }()
+ req := base.RestyClient.R().SetContext(ctx)
req.SetQueryParam("access_token", d.AccessToken)
if callback != nil {
callback(req)
@@ -107,34 +124,59 @@ func (d *BaiduNetdisk) request(furl string, method string, callback base.ReqCall
}
res, err := req.Execute(method, furl)
if err != nil {
+ if ctx.Err() != nil {
+ return retry.Unrecoverable(ctx.Err())
+ }
return err
}
log.Debugf("[baidu_netdisk] req: %s, resp: %s", furl, res.String())
- errno := utils.Json.Get(res.Body(), "errno").ToInt()
- if errno != 0 {
- if utils.SliceContains([]int{111, -6, 20016, 20017}, errno) {
- log.Info("[baidu_netdisk] refreshing baidu_netdisk token.")
- err2 := d.refreshToken()
- if err2 != nil {
- return retry.Unrecoverable(err2)
- }
- }
-
- if errno == 31023 && d.DownloadAPI == "crack_video" {
- result = res.Body()
- return nil
+ apiErr := parseAPIError(furl, res.Body())
+ if apiErr == nil {
+ result = outcomeSuccess
+ body = res.Body()
+ return nil
+ }
+ errno := apiErr.Errno
+ if apiErr.RateLimited() {
+ result = outcomeRateLimited
+ return apiErr
+ }
+ // Baidu answered without throttling: as far as the breaker is
+ // concerned this is a success, even if the call itself failed.
+ result = outcomeSuccess
+ if apiErr.NotExist() {
+ // permanent; callers such as Remove decide whether that is fine
+ return retry.Unrecoverable(apiErr)
+ }
+ // NOTE: per the filemanager docs 111 means "an async task is running",
+ // but it has always been treated as token expiry here; kept unchanged.
+ // 20016/20017 are additional token-refresh errnos from upstream (#3232).
+ if utils.SliceContains([]int{111, -6, 20016, 20017}, errno) {
+ log.Info("[baidu_netdisk] refreshing baidu_netdisk token.")
+ err2 := d.refreshToken()
+ if err2 != nil {
+ return retry.Unrecoverable(err2)
}
+ }
- return fmt.Errorf("req: [%s] ,errno: %d, refer to https://pan.baidu.com/union/doc/", furl, errno)
+ if errno == 31023 && d.DownloadAPI == "crack_video" {
+ body = res.Body()
+ return nil
}
- result = res.Body()
- return nil
+
+ return apiErr
},
+ retry.Context(ctx),
retry.LastErrorOnly(true),
retry.Attempts(3),
retry.Delay(time.Second),
- retry.DelayType(retry.BackOffDelay))
- return result, err
+ retry.DelayType(func(n uint, err error, config *retry.Config) time.Duration {
+ if e, ok := asAPIError(err); ok && e.RateLimited() {
+ return 0 // the shared cooldown provides the delay
+ }
+ return retry.BackOffDelay(n, err, config)
+ }))
+ return body, err
}
func (d *BaiduNetdisk) get(pathname string, params map[string]string, resp interface{}) ([]byte, error) {
@@ -144,7 +186,11 @@ func (d *BaiduNetdisk) get(pathname string, params map[string]string, resp inter
}
func (d *BaiduNetdisk) postForm(pathname string, params map[string]string, form map[string]string, resp interface{}) ([]byte, error) {
- return d.request("https://pan.baidu.com/rest/2.0"+pathname, http.MethodPost, func(req *resty.Request) {
+ return d.postFormCtx(context.Background(), pathname, params, form, resp)
+}
+
+func (d *BaiduNetdisk) postFormCtx(ctx context.Context, pathname string, params map[string]string, form map[string]string, resp interface{}) ([]byte, error) {
+ return d.requestCtx(ctx, "https://pan.baidu.com/rest/2.0"+pathname, http.MethodPost, func(req *resty.Request) {
req.SetQueryParams(params)
req.SetFormData(form)
}, resp)
@@ -276,13 +322,13 @@ func (d *BaiduNetdisk) linkCrackVideo(file model.Obj, _ model.LinkArgs) (*model.
}, nil
}
-func (d *BaiduNetdisk) manage(opera string, filelist any) ([]byte, error) {
+func (d *BaiduNetdisk) manage(ctx context.Context, opera string, filelist any) ([]byte, error) {
params := map[string]string{
"method": "filemanager",
"opera": opera,
}
marshal, _ := utils.Json.MarshalToString(filelist)
- return d.postForm("/xpan/file", params, map[string]string{
+ return d.postFormCtx(ctx, "/xpan/file", params, map[string]string{
"async": "0",
"filelist": marshal,
"ondup": "fail",
diff --git a/server/s3/backend.go b/server/s3/backend.go
index de6413ce9b..223cfe5a6b 100644
--- a/server/s3/backend.go
+++ b/server/s3/backend.go
@@ -341,9 +341,13 @@ func (b *s3Backend) DeleteMulti(ctx context.Context, bucketName string, objects
for _, object := range objects {
if err := b.deleteObject(ctx, bucketName, object); err != nil {
log.Errorf("delete object failed: %v", err)
+ code := gofakes3.ErrInternal
+ if gofakes3.HasErrorCode(err, gofakes3.ErrSlowDown) {
+ code = gofakes3.ErrSlowDown
+ }
result.Error = append(result.Error, gofakes3.ErrorResult{
- Code: gofakes3.ErrInternal,
- Message: gofakes3.ErrInternal.Message(),
+ Code: code,
+ Message: code.Message(),
Key: object,
})
} else {
@@ -374,10 +378,15 @@ func (b *s3Backend) deleteObject(ctx context.Context, bucketName, objectName str
// S3 does not report an error when attemping to delete a key that does not exist, so
// we need to skip IsNotExist errors.
if _, err := fs.Get(context.WithValue(ctx, conf.MetaKey, fmeta), fp, &fs.GetArgs{}); err != nil && !errs.IsObjectNotFound(err) {
- return err
+ return mapBackendError(err)
}
- fs.Remove(ctx, fp)
+ // Report failures (op.Remove already treats a missing object as success) so
+ // clients retry instead of leaving the object behind; rate limiting becomes
+ // 503 SlowDown, which S3 clients retry with backoff.
+ if err := fs.Remove(ctx, fp); err != nil {
+ return mapBackendError(err)
+ }
return nil
}
diff --git a/server/s3/delete_test.go b/server/s3/delete_test.go
new file mode 100644
index 0000000000..56d942602b
--- /dev/null
+++ b/server/s3/delete_test.go
@@ -0,0 +1,164 @@
+package s3
+
+import (
+ "context"
+ "encoding/json"
+ "encoding/xml"
+ "errors"
+ "io"
+ "net/http"
+ "net/http/httptest"
+ "os"
+ "path/filepath"
+ "strings"
+ "testing"
+
+ "github.com/OpenListTeam/OpenList/v4/drivers/local"
+ "github.com/OpenListTeam/OpenList/v4/internal/conf"
+ "github.com/OpenListTeam/OpenList/v4/internal/driver"
+ "github.com/OpenListTeam/OpenList/v4/internal/errs"
+ "github.com/OpenListTeam/OpenList/v4/internal/model"
+ "github.com/OpenListTeam/OpenList/v4/internal/op"
+ "github.com/OpenListTeam/gofakes3"
+)
+
+const removeErrDriverName = "S3RemoveErrLocal"
+
+// removeErr is the error the fixture driver's Remove returns (nil = delete normally).
+var removeErr error
+
+type removeErrDriver struct{ local.Local }
+
+func (d *removeErrDriver) Config() driver.Config {
+ c := d.Local.Config()
+ c.Name = removeErrDriverName
+ return c
+}
+
+func (d *removeErrDriver) Remove(ctx context.Context, obj model.Obj) error {
+ if removeErr != nil {
+ return removeErr
+ }
+ return d.Local.Remove(ctx, obj)
+}
+
+func setupDeleteServer(t *testing.T) (*httptest.Server, string) {
+ t.Helper()
+ ctx := context.Background()
+ op.RegisterDriver(func() driver.Driver { return &removeErrDriver{} })
+ root := t.TempDir()
+ addition, _ := json.Marshal(struct {
+ RootFolderPath string `json:"root_folder_path"`
+ }{RootFolderPath: root})
+ mount := "/" + sanitizeTestName(t.Name())
+ id, err := op.CreateStorage(ctx, model.Storage{Driver: removeErrDriverName, MountPath: mount, Addition: string(addition)})
+ if err != nil {
+ t.Fatal(err)
+ }
+ t.Cleanup(func() { _ = op.DeleteStorageById(ctx, id) })
+ if err := op.SaveSettingItem(&model.SettingItem{Key: conf.S3Buckets, Value: `[{"name":"del","path":"` + mount + `"}]`}); err != nil {
+ t.Fatal(err)
+ }
+ srv := httptest.NewServer(gofakes3.New(newBackend(), gofakes3.WithoutVersioning()).Server())
+ t.Cleanup(srv.Close)
+ t.Cleanup(func() { removeErr = nil })
+ return srv, root
+}
+
+func doReq(t *testing.T, method, url, body string) (*http.Response, string) {
+ t.Helper()
+ req, _ := http.NewRequest(method, url, strings.NewReader(body))
+ resp, err := http.DefaultClient.Do(req)
+ if err != nil {
+ t.Fatal(err)
+ }
+ defer resp.Body.Close()
+ b, _ := io.ReadAll(resp.Body)
+ return resp, string(b)
+}
+
+func TestDeleteObjectReportsRemoveErrors(t *testing.T) {
+ srv, root := setupDeleteServer(t)
+ write := func(name string) string {
+ p := filepath.Join(root, name)
+ if err := os.WriteFile(p, []byte("x"), 0o600); err != nil {
+ t.Fatal(err)
+ }
+ return p
+ }
+ exists := func(p string) bool { _, err := os.Stat(p); return err == nil }
+
+ cases := []struct {
+ name string
+ err error
+ wantStatus int
+ wantCode gofakes3.ErrorCode
+ wantGone bool
+ }{
+ {"rate limited", errs.NewErr(errs.TemporaryCapacity, "baidu 31034"), http.StatusServiceUnavailable, gofakes3.ErrSlowDown, false},
+ {"other error", errors.New("baidu errno 12"), http.StatusInternalServerError, gofakes3.ErrInternal, false},
+ {"success", nil, http.StatusNoContent, "", true},
+ }
+ for i, c := range cases {
+ t.Run(c.name, func(t *testing.T) {
+ name := "obj" + string(rune('a'+i))
+ p := write(name)
+ removeErr = c.err
+ resp, body := doReq(t, http.MethodDelete, srv.URL+"/del/"+name, "")
+ if resp.StatusCode != c.wantStatus {
+ t.Fatalf("status = %d, want %d (%s)", resp.StatusCode, c.wantStatus, body)
+ }
+ if c.wantCode != "" && !strings.Contains(body, ""+string(c.wantCode)+"") {
+ t.Fatalf("body %q lacks code %s", body, c.wantCode)
+ }
+ if exists(p) == c.wantGone {
+ t.Fatalf("file exists=%v, want gone=%v", exists(p), c.wantGone)
+ }
+ })
+ }
+
+ t.Run("missing key is success", func(t *testing.T) {
+ removeErr = errors.New("must not be called")
+ resp, body := doReq(t, http.MethodDelete, srv.URL+"/del/never-existed", "")
+ if resp.StatusCode != http.StatusNoContent {
+ t.Fatalf("status = %d (%s)", resp.StatusCode, body)
+ }
+ })
+}
+
+func TestDeleteMultiReportsPerKeyErrors(t *testing.T) {
+ srv, root := setupDeleteServer(t)
+ for _, n := range []string{"k1", "k2"} {
+ if err := os.WriteFile(filepath.Join(root, n), []byte("x"), 0o600); err != nil {
+ t.Fatal(err)
+ }
+ }
+ removeErr = errs.NewErr(errs.TemporaryCapacity, "baidu 31034")
+ resp, body := doReq(t, http.MethodPost, srv.URL+"/del?delete",
+ ``)
+ if resp.StatusCode != http.StatusOK {
+ t.Fatalf("status = %d (%s)", resp.StatusCode, body)
+ }
+ var out struct {
+ Deleted []struct{ Key string } `xml:"Deleted"`
+ Error []struct {
+ Key string
+ Code string
+ } `xml:"Error"`
+ }
+ if err := xml.Unmarshal([]byte(body), &out); err != nil {
+ t.Fatal(err)
+ }
+ if len(out.Error) != 1 || out.Error[0].Key != "k1" || out.Error[0].Code != string(gofakes3.ErrSlowDown) {
+ t.Fatalf("errors = %+v, want k1 SlowDown (%s)", out.Error, body)
+ }
+ if len(out.Deleted) != 1 || out.Deleted[0].Key != "gone" {
+ t.Fatalf("deleted = %+v, want [gone]", out.Deleted)
+ }
+
+ removeErr = errors.New("boom")
+ _, body = doReq(t, http.MethodPost, srv.URL+"/del?delete", ``)
+ if !strings.Contains(body, "InternalError") {
+ t.Fatalf("want InternalError for k2: %s", body)
+ }
+}