zoho: log throttling once per episode at NOTICE

A 429 stall was only visible as a DEBUG pacer line, so without -vv
rclone appeared to hang for 2-5 minutes. In one night's batch logs 17
job starts produced only 4 completions because the silent stalls
looked like hangs and the jobs kept getting killed, re-triggering the
throttle.

Log the first 429 of each throttle episode at NOTICE with the server
message and the wait time. An episode ends when a request succeeds
after the penalty window; retries within an episode stay at DEBUG via
the existing pacer logging. State is two atomics behind a pointer on
Fs, so shallow Fs copies share it and concurrent checkers are safe.

See #9570

backend/zoho/throttle_test.go

@@ -3,6 +3,7 @@ package zoho
 import (
 	"context"
 	"net/http"
+	"sync"
 	"testing"
 	"time"
 
@@ -11,10 +12,13 @@ import (
 	"github.com/stretchr/testify/require"
 )
 
-// newTestFs returns a bare *Fs so the 429/retry logic in shouldRetry can be
-// exercised in isolation. No network or pacer is involved.
+// newTestFs returns a bare *Fs carrying just the throttle state shouldRetry
+// touches, armed exactly as NewFs arms it. No network or pacer is involved, so
+// the 429/retry logic can be exercised in isolation.
 func newTestFs() *Fs {
-	return &Fs{}
+	f := &Fs{throttle: &throttleState{}}
+	f.throttle.progress.Store(true)
+	return f
 }
 
 func TestShouldRetry(t *testing.T) {
@@ -78,3 +82,52 @@ func TestShouldRetry(t *testing.T) {
 		assert.False(t, retry)
 	})
 }
+
+// TestThrottleEpisode covers the once-per-episode logging state machine that
+// logThrottle/shouldRetry drive through throttleState, without sleeping: the
+// penalty window is moved by hand instead of waited out.
+func TestThrottleEpisode(t *testing.T) {
+	ctx := context.Background()
+	f := newTestFs() // progress armed: the first 429 would log at NOTICE
+	resp429 := &http.Response{StatusCode: 429, Header: http.Header{"Retry-After": {"1"}}}
+	respOK := &http.Response{StatusCode: 200}
+
+	// The first 429 consumes the armed flag (logs once) and opens a penalty window.
+	_, _ = f.shouldRetry(ctx, resp429, assert.AnError)
+	assert.False(t, f.throttle.progress.Load(), "first 429 disarms progress")
+
+	// A success still inside the penalty window must not re-arm (would let a
+	// burst of in-flight successes start a fresh episode too early).
+	_, _ = f.shouldRetry(ctx, respOK, nil)
+	assert.False(t, f.throttle.progress.Load(), "success during penalty window does not re-arm")
+
+	// Once the penalty window has elapsed, a success re-arms for the next episode.
+	f.throttle.penaltyUntilNano.Store(time.Now().Add(-time.Second).UnixNano())
+	_, _ = f.shouldRetry(ctx, respOK, nil)
+	assert.True(t, f.throttle.progress.Load(), "success after penalty window re-arms")
+}
+
+// TestThrottleStateConcurrent drives the lock-free throttleState (shared across
+// the shallow Fs copy) from many goroutines. Like the registry test, `go test
+// -race` is the real assertion: it would flag any non-atomic access. The final
+// flag is interleaving-dependent, so it is deliberately not asserted.
+func TestThrottleStateConcurrent(t *testing.T) {
+	ctx := context.Background()
+	f := newTestFs()
+	resp429 := &http.Response{StatusCode: 429, Header: http.Header{"Retry-After": {"1"}}}
+	respOK := &http.Response{StatusCode: 200}
+
+	var wg sync.WaitGroup
+	for i := range 64 {
+		wg.Add(1)
+		go func(i int) {
+			defer wg.Done()
+			if i%2 == 0 {
+				_, _ = f.shouldRetry(ctx, resp429, assert.AnError)
+			} else {
+				_, _ = f.shouldRetry(ctx, respOK, nil)
+			}
+		}(i)
+	}
+	wg.Wait()
+}

backend/zoho/zoho.go

@@ -12,6 +12,7 @@ import (
 	"path"
 	"strconv"
 	"strings"
+	"sync/atomic"
 	"time"
 
 	"github.com/google/uuid"
@@ -291,6 +292,17 @@ type Fs struct {
 	uploadsrv   *rest.Client       // the connection to the upload server
 	dirCache    *dircache.DirCache // Map of directory path to directory id
 	pacer       *fs.Pacer          // pacer for API calls
+	throttle    *throttleState     // once-per-episode 429 logging state (shared across Fs copies)
+}
+
+// throttleState tracks 429 throttling for once-per-episode logging (see
+// logThrottle). It sits behind a pointer so the "assume it is a file" shallow
+// copy of Fs shares one state; both fields are atomic, so it is lock-free. An
+// "episode" is a run of 429s with no recovery (a success after the penalty
+// window clears) between them.
+type throttleState struct {
+	penaltyUntilNano atomic.Int64 // unix-nanos the current 429 penalty should clear
+	progress         atomic.Bool  // a call succeeded after the last penalty window
 }
 
 // Object describes a Zoho WorkDrive object
@@ -421,17 +433,41 @@ var retryErrorCodes = []int{
 	509, // Bandwidth Limit Exceeded
 }
 
+// logThrottle logs the first 429 in a throttling episode at NOTICE level.
+//
+// The err value contains the server response body. Further 429s are not logged
+// until shouldRetry observes a real recovery, which prevents sustained
+// throttling from flooding the log. The pacer still logs every retry at DEBUG.
+// Using recovery instead of a fixed time window works with both Zoho penalty
+// regimes.
+func (f *Fs) logThrottle(wait time.Duration, err error) {
+	newBurst := f.throttle.progress.Swap(false)
+	f.throttle.penaltyUntilNano.Store(time.Now().Add(wait).UnixNano())
+	secs := int(wait / time.Second)
+	if newBurst {
+		fs.Logf(f, "Too many requests: Trying again in %d seconds. %v", secs, err)
+	}
+}
+
 // shouldRetry reports whether the given resp and err deserve to be retried.
 //
 // A 429 is honoured via the Retry-After header (falling back to 60s plus a
-// margin); expired OAuth tokens are retried, missing OAuth scopes abort, and
-// standard HTTP retry conditions are also handled.
+// margin) and starts or continues a throttling episode; expired OAuth tokens
+// are retried, missing OAuth scopes abort, and standard HTTP retry conditions
+// are also handled.
 //
 // Returns whether to retry, and the err as a convenience.
 func (f *Fs) shouldRetry(ctx context.Context, resp *http.Response, err error) (bool, error) {
 	if fserrors.ContextError(ctx, &err) {
 		return false, err
 	}
+	if err == nil && resp != nil && resp.StatusCode < 400 {
+		// Treat as recovered only after the latest 429 wait ends, so in-flight
+		// successes don't start a new throttling episode too early.
+		if time.Now().UnixNano() > f.throttle.penaltyUntilNano.Load() {
+			f.throttle.progress.Store(true)
+		}
+	}
 	authRetry := false
 
 	// Bail out early if we are missing OAuth Scopes.
@@ -454,10 +490,12 @@ func (f *Fs) shouldRetry(ctx context.Context, resp *http.Response, err error) (b
 				fs.Logf(f, "Failed to parse Retry-After: %q: %v", values[0], parseErr)
 			} else {
 				wait := time.Duration(retryAfter)*time.Second + retryAfterMargin
+				f.logThrottle(wait, err)
 				return true, pacer.RetryAfterError(err, wait)
 			}
 		}
 		wait := 60*time.Second + retryAfterMargin
+		f.logThrottle(wait, err)
 		return true, pacer.RetryAfterError(err, wait)
 	}
 	return authRetry || fserrors.ShouldRetry(err) || fserrors.ShouldRetryHTTP(resp, retryErrorCodes), err
@@ -595,7 +633,10 @@ func NewFs(ctx context.Context, name, root string, m configmap.Mapper) (fs.Fs, e
 		downloadsrv: rest.NewClient(oAuthClient).SetRoot(downloadURL),
 		uploadsrv:   rest.NewClient(oAuthClient).SetRoot(uploadURL),
 		pacer:       fs.NewPacer(ctx, pacer.NewGoogleDrive(pacer.MinSleep(pacerMinSleep), pacer.Burst(pacerBurst))),
+		throttle:    &throttleState{},
 	}
+	// Arm progress so the very first 429 is logged at NOTICE.
+	f.throttle.progress.Store(true)
 	f.features = (&fs.Features{
 		CanHaveEmptyDirectories: true,
 	}).Fill(ctx, f)