fix(gitsync): concurrent refresh, decoupled timeout, and no-token backoff (#23004)

## Problem The gitsync worker polls every 10s and refreshes up to 50 stale `chat_diff_status` rows **sequentially**, sharing a single 10-second context timeout. With 50 rows × 1–3 HTTP calls each, the timeout is exhausted quickly, causing cascading `context deadline exceeded` errors. Rows with no linked OAuth token (`ErrNoTokenAvailable`) fail fast but recur every 120s, wasting batch capacity. ## Solution Three targeted fixes: ### 1. Concurrent refresh processing `Refresher.Refresh()` now launches goroutines bounded by a semaphore (`defaultConcurrency = 10`). Provider/token resolution remains sequential (fast DB lookups); only the HTTP calls run in parallel. Per-group rate-limit detection uses `atomic.Pointer[RateLimitError]` with best-effort skip of remaining rows — a rate-limit hit on one provider doesn't stall requests to other providers. ### 2. Decoupled tick timeout New `defaultTickTimeout = 30s`, separate from `defaultInterval = 10s`. The `tick()` method uses `tickTimeout` for its context deadline, giving concurrent HTTP calls enough headroom to complete without stalling the next polling cycle. ### 3. Longer backoff for no-token errors New `NoTokenBackoff = 10 * time.Minute` (exported). When `errors.Is(err, ErrNoTokenAvailable)`, the worker applies a 10-minute backoff instead of `DiffStatusTTL` (2 minutes). Retrying every 2 minutes is pointless until the user manually links their external auth account. ## Design decisions - Both `NewRefresher` and `NewWorker` accept variadic option functions (`RefresherOption`, `WorkerOption`) for backward compatibility — existing callers in `coderd/coderd.go` need no changes. - `WithConcurrency(n)` and `WithTickTimeout(d)` are available for tests and future tuning. - Added `resolvedGroup` struct to cleanly separate the pre-resolution phase from the concurrent execution phase. ## Testing - **`TestRefresher_RateLimitSkipsRemainingInGroup`** — rewritten to be goroutine-order-independent (verifies aggregate counts instead of per-index results). - **`TestRefresher_ConcurrentProcessing`** — new test using a gate channel to prove N goroutines enter `FetchPullRequestStatus` simultaneously. - **`TestWorker_RefresherError_BacksOffRow`** — rewritten to use branch-name-based failure determination instead of non-deterministic `callCount`. - **`TestWorker_NoTokenBackoff`** — new test verifying `ErrNoTokenAvailable` triggers 10-minute backoff. - All tests pass under `-race -count=3`.
2026-09-24 15:04:27 +08:00 · 2026-03-12 18:08:06 +00:00
parent fc9e04da67
commit b1e80e6f3a
4 changed files with 342 additions and 105 deletions
@@ -5,6 +5,8 @@ import (
 	"database/sql"
 	"errors"
 	"fmt"
+	"sync"
+	"sync/atomic"
 	"time"

 	"github.com/google/uuid"
@@ -20,6 +22,10 @@ const (
 	// DiffStatusTTL is how long a successfully refreshed
 	// diff status remains fresh before becoming stale again.
 	DiffStatusTTL = 120 * time.Second
+
+	// defaultConcurrency is the maximum number of HTTP calls
+	// made in parallel during a single Refresh batch.
+	defaultConcurrency = 10
 )

 // ProviderResolver maps a git remote origin to the gitprovider
@@ -28,6 +34,10 @@ type ProviderResolver func(origin string) gitprovider.Provider

 var ErrNoTokenAvailable error = errors.New("no token available")

+// ErrRateLimitSkipped indicates that a row was skipped because
+// a prior request in the same group hit a rate limit.
+var ErrRateLimitSkipped error = errors.New("skipped due to rate limit")
+
 // TokenResolver obtains the user's git access token for a given
 // remote origin. Should return nil if no token is available, in
 // which case ErrNoTokenAvailable will be returned.
@@ -37,14 +47,28 @@ type TokenResolver func(
 	origin string,
 ) (*string, error)

+// RefresherOption configures a Refresher.
+type RefresherOption func(*Refresher)
+
+// WithConcurrency sets the maximum number of concurrent HTTP
+// calls per Refresh batch. Defaults to defaultConcurrency.
+func WithConcurrency(n int) RefresherOption {
+	return func(r *Refresher) {
+		if n > 0 {
+			r.concurrency = n
+		}
+	}
+}
+
 // Refresher contains the stateless business logic for fetching
 // fresh PR data from a git provider given a stale
 // database.ChatDiffStatus row.
 type Refresher struct {
-	providers ProviderResolver
-	tokens    TokenResolver
-	logger    slog.Logger
-	clock     quartz.Clock
+	providers   ProviderResolver
+	tokens      TokenResolver
+	logger      slog.Logger
+	clock       quartz.Clock
+	concurrency int
 }

 // NewRefresher creates a Refresher with the given dependency
@@ -54,13 +78,19 @@ func NewRefresher(
 	tokens TokenResolver,
 	logger slog.Logger,
 	clock quartz.Clock,
+	opts ...RefresherOption,
 ) *Refresher {
-	return &Refresher{
-		providers: providers,
-		tokens:    tokens,
-		logger:    logger,
-		clock:     clock,
+	r := &Refresher{
+		providers:   providers,
+		tokens:      tokens,
+		logger:      logger,
+		clock:       clock,
+		concurrency: defaultConcurrency,
 	}
+	for _, o := range opts {
+		o(r)
+	}
+	return r
 }

 // RefreshRequest pairs a stale row with the chat owner who
@@ -87,10 +117,21 @@ type groupKey struct {
 	origin  string
 }

+// resolvedGroup holds the pre-resolved provider and token for
+// a group of requests that share the same (owner, origin).
+type resolvedGroup struct {
+	provider gitprovider.Provider
+	token    string
+	indices  []int
+}
+
 // Refresh fetches fresh PR data for a batch of stale rows.
 // Rows are grouped internally by (ownerID, origin) so that
-// provider and token resolution happen once per group. A
-// top-level error is returned only when the entire batch
+// provider and token resolution happen once per group. HTTP
+// calls within and across groups run concurrently, bounded by
+// the Refresher's concurrency limit.
+//
+// A top-level error is returned only when the entire batch
 // fails catastrophically. Per-row outcomes are in the
 // returned RefreshResult slice (one per input request, same
 // order).
@@ -113,6 +154,10 @@ func (r *Refresher) Refresh(
 		groups[key] = append(groups[key], i)
 	}

+	// Pre-resolve providers and tokens sequentially. This is
+	// fast (DB + in-memory config lookups) and avoids
+	// duplicate resolution for rows in the same group.
+	var resolved []resolvedGroup
 	for key, indices := range groups {
 		provider := r.providers(key.origin)
 		if provider == nil {
@@ -135,35 +180,82 @@ func (r *Refresher) Refresh(
 			}
 			continue
 		}
-		// This is technically unnecessary but kept here as a future molly-guard.
-		if token == nil {
-			continue
-		}

-		for i, idx := range indices {
-			req := requests[idx]
-			params, err := r.refreshOne(ctx, provider, *token, req.Row)
-			results[idx] = RefreshResult{Request: req, Params: params, Error: err}
+		resolved = append(resolved, resolvedGroup{
+			provider: provider,
+			token:    *token,
+			indices:  indices,
+		})
+	}

-			// If rate-limited, skip remaining rows in this group.
-			var rlErr *gitprovider.RateLimitError
-			if errors.As(err, &rlErr) {
-				for _, remaining := range indices[i+1:] {
-					results[remaining] = RefreshResult{
-						Request: requests[remaining],
-						Error:   fmt.Errorf("skipped: %w", rlErr),
+	// Process all HTTP calls concurrently with a shared
+	// semaphore. Each group tracks rate-limit errors
+	// independently so that a limit hit on one provider
+	// doesn't stall requests to other providers.
+	sem := make(chan struct{}, r.concurrency)
+	var wg sync.WaitGroup
+
+	for _, grp := range resolved {
+		var rateLimitErr atomic.Pointer[gitprovider.RateLimitError]
+
+		for _, idx := range grp.indices {
+			wg.Add(1)
+			go func() {
+				defer wg.Done()
+
+				// Best-effort rate-limit check before acquiring
+				// the semaphore to avoid unnecessary blocking.
+				if rl := rateLimitErr.Load(); rl != nil {
+					results[idx] = RefreshResult{
+						Request: requests[idx],
+						Error:   fmt.Errorf("%w: %w", ErrRateLimitSkipped, rl),
 					}
+					return
 				}
-				break
-			}
+
+				// Acquire semaphore slot.
+				select {
+				case sem <- struct{}{}:
+					defer func() { <-sem }()
+				case <-ctx.Done():
+					results[idx] = RefreshResult{
+						Request: requests[idx],
+						Error:   ctx.Err(),
+					}
+					return
+				}
+
+				// Best-effort rate-limit check after acquiring
+				// in case it was set while we waited.
+				if rl := rateLimitErr.Load(); rl != nil {
+					results[idx] = RefreshResult{
+						Request: requests[idx],
+						Error:   fmt.Errorf("%w: %w", ErrRateLimitSkipped, rl),
+					}
+					return
+				}
+
+				params, err := r.refreshOne(ctx, grp.provider, grp.token, requests[idx].Row)
+				results[idx] = RefreshResult{
+					Request: requests[idx],
+					Params:  params,
+					Error:   err,
+				}
+
+				var rlErr *gitprovider.RateLimitError
+				if errors.As(err, &rlErr) {
+					rateLimitErr.Store(rlErr)
+				}
+			}()
 		}
 	}

+	wg.Wait()
 	return results, nil
 }

 // refreshOne processes a single row using an already-resolved
-// provider and token. This is the old Refresh logic, unchanged.
+// provider and token.
 func (r *Refresher) refreshOne(
 	ctx context.Context,
 	provider gitprovider.Provider,
@@ -560,41 +560,16 @@ func TestRefresher_RateLimitSkipsRemainingInGroup(t *testing.T) {

 	mp := &mockProvider{
 		parsePullRequestURL: func(raw string) (gitprovider.PRRef, bool) {
-			var num int
-			switch {
-			case strings.HasSuffix(raw, "/pull/1"):
-				num = 1
-			case strings.HasSuffix(raw, "/pull/2"):
-				num = 2
-			case strings.HasSuffix(raw, "/pull/3"):
-				num = 3
-			default:
-				return gitprovider.PRRef{}, false
-			}
-			return gitprovider.PRRef{Owner: "org", Repo: "repo", Number: num}, true
+			return gitprovider.PRRef{Owner: "org", Repo: "repo", Number: 1}, raw != ""
 		},
-		fetchPullRequestStatus: func(_ context.Context, _ string, ref gitprovider.PRRef) (*gitprovider.PRStatus, error) {
-			call := callCount.Add(1)
-			switch call {
-			case 1:
-				// First call succeeds.
-				return &gitprovider.PRStatus{
-					State: gitprovider.PRStateOpen,
-					DiffStats: gitprovider.DiffStats{
-						Additions:    5,
-						Deletions:    2,
-						ChangedFiles: 1,
-					},
-				}, nil
-			case 2:
-				// Second call hits rate limit.
-				return nil, &gitprovider.RateLimitError{
-					RetryAfter: time.Now().Add(60 * time.Second),
-				}
-			default:
-				// Third call should never happen.
-				t.Fatal("FetchPullRequestStatus called more than 2 times")
-				return nil, nil
+		fetchPullRequestStatus: func(_ context.Context, _ string, _ gitprovider.PRRef) (*gitprovider.PRStatus, error) {
+			// Every call returns a rate limit error. With
+			// concurrency=1 the first goroutine to acquire the
+			// semaphore makes the only real call; remaining
+			// goroutines see the flag and skip.
+			callCount.Add(1)
+			return nil, &gitprovider.RateLimitError{
+				RetryAfter: time.Now().Add(60 * time.Second),
 			}
 		},
 	}
@@ -604,7 +579,9 @@ func TestRefresher_RateLimitSkipsRemainingInGroup(t *testing.T) {
 		return ptr.Ref("test-token"), nil
 	}

-	r := gitsync.NewRefresher(providers, tokens, slogtest.Make(t, nil), quartz.NewReal())
+	// Concurrency=1 ensures sequential semaphore acquisition so
+	// the rate-limit flag is always visible to later goroutines.
+	r := gitsync.NewRefresher(providers, tokens, slogtest.Make(t, nil), quartz.NewReal(), gitsync.WithConcurrency(1))

 	ownerID := uuid.New()
 	origin := "https://github.com/org/repo"
@@ -643,26 +620,31 @@ func TestRefresher_RateLimitSkipsRemainingInGroup(t *testing.T) {
 	require.NoError(t, err)
 	require.Len(t, results, 3)

-	// Row 0: success.
-	assert.NoError(t, results[0].Error)
-	assert.NotNil(t, results[0].Params)
+	// With concurrency=1, the first goroutine to acquire the
+	// semaphore makes the only API call (which rate-limits).
+	// The remaining goroutines see the rate-limit flag and
+	// skip. Goroutine scheduling order is non-deterministic,
+	// so we verify aggregate counts rather than per-index
+	// results.
+	var directCount, skippedCount int
+	for _, res := range results {
+		require.Error(t, res.Error)
+		var rlErr *gitprovider.RateLimitError
+		require.True(t, errors.As(res.Error, &rlErr),
+			"every result should wrap *RateLimitError")
+		if errors.Is(res.Error, gitsync.ErrRateLimitSkipped) {
+			skippedCount++
+		} else {
+			directCount++
+		}
+	}

-	// Row 1: rate-limited.
-	require.Error(t, results[1].Error)
-	var rlErr1 *gitprovider.RateLimitError
-	assert.True(t, errors.As(results[1].Error, &rlErr1),
-		"result[1] error should be *RateLimitError")
-
-	// Row 2: skipped due to rate limit.
-	require.Error(t, results[2].Error)
-	var rlErr2 *gitprovider.RateLimitError
-	assert.True(t, errors.As(results[2].Error, &rlErr2),
-		"result[2] error should wrap *RateLimitError")
-	assert.Contains(t, results[2].Error.Error(), "skipped")
-
-	// Provider should have been called exactly twice.
-	assert.Equal(t, int32(2), callCount.Load(),
-		"FetchPullRequestStatus should be called exactly 2 times")
+	assert.Equal(t, 1, directCount,
+		"exactly one row should be directly rate-limited")
+	assert.Equal(t, 2, skippedCount,
+		"two rows should be skipped due to rate limit")
+	assert.Equal(t, int32(1), callCount.Load(),
+		"FetchPullRequestStatus should be called exactly once")
 }

 func TestRefresher_CorrectTokenPerOrigin(t *testing.T) {
@@ -773,3 +755,69 @@ func TestRefresher_CorrectTokenPerOrigin(t *testing.T) {
 	assert.Equal(t, int32(2), tokenCalls.Load(),
 		"TokenResolver should be called once per (owner, origin) group")
 }
+
+func TestRefresher_ConcurrentProcessing(t *testing.T) {
+	t.Parallel()
+
+	const numRows = 3
+
+	// gate blocks all goroutines until numRows goroutines have
+	// entered FetchPullRequestStatus, proving they run concurrently.
+	gate := make(chan struct{})
+	var entered atomic.Int32
+
+	mp := &mockProvider{
+		parsePullRequestURL: func(raw string) (gitprovider.PRRef, bool) {
+			return gitprovider.PRRef{Owner: "org", Repo: "repo", Number: 1}, true
+		},
+		fetchPullRequestStatus: func(_ context.Context, _ string, _ gitprovider.PRRef) (*gitprovider.PRStatus, error) {
+			if entered.Add(1) == numRows {
+				close(gate)
+			}
+			// Block until all goroutines have entered.
+			<-gate
+			return &gitprovider.PRStatus{State: gitprovider.PRStateOpen}, nil
+		},
+	}
+
+	providers := func(_ string) gitprovider.Provider { return mp }
+	tokens := func(_ context.Context, _ uuid.UUID, _ string) (*string, error) {
+		return ptr.Ref("test-token"), nil
+	}
+
+	// Concurrency must be >= numRows so all goroutines can enter
+	// simultaneously.
+	r := gitsync.NewRefresher(providers, tokens, slogtest.Make(t, nil), quartz.NewReal(), gitsync.WithConcurrency(numRows))
+
+	ownerID := uuid.New()
+	origin := "https://github.com/org/repo"
+
+	requests := make([]gitsync.RefreshRequest, numRows)
+	for i := range requests {
+		requests[i] = gitsync.RefreshRequest{
+			Row: database.ChatDiffStatus{
+				ChatID:          uuid.New(),
+				Url:             sql.NullString{String: fmt.Sprintf("https://github.com/org/repo/pull/%d", i+1), Valid: true},
+				GitRemoteOrigin: origin,
+				GitBranch:       fmt.Sprintf("feat-%d", i+1),
+			},
+			OwnerID: ownerID,
+		}
+	}
+
+	results, err := r.Refresh(context.Background(), requests)
+	require.NoError(t, err)
+	require.Len(t, results, numRows)
+
+	for i, res := range results {
+		if res.Error != nil {
+			t.Logf("result[%d] error: %v", i, res.Error)
+		}
+		assert.NoError(t, res.Error, "result[%d]", i)
+		assert.NotNil(t, res.Params, "result[%d]", i)
+	}
+
+	// All numRows goroutines entered FetchPullRequestStatus
+	// concurrently.
+	assert.Equal(t, int32(numRows), entered.Load())
+}
@@ -3,6 +3,7 @@ package gitsync
 import (
 	"context"
 	"database/sql"
+	"errors"
 	"time"

 	"github.com/google/uuid"
@@ -20,6 +21,17 @@ const (

 	// defaultInterval is the polling interval between ticks.
 	defaultInterval = 10 * time.Second
+
+	// defaultTickTimeout is the maximum time a single tick may
+	// run. Decoupled from the polling interval so that a batch
+	// of concurrent HTTP calls has enough headroom to complete.
+	defaultTickTimeout = 30 * time.Second
+
+	// NoTokenBackoff is the backoff duration applied to rows
+	// whose owner has no linked external-auth token. Much longer
+	// than DiffStatusTTL because the user must manually link
+	// their account before retrying is useful.
+	NoTokenBackoff = 10 * time.Minute
 )

 // Store is the narrow DB interface the Worker needs.
@@ -54,9 +66,22 @@ type Worker struct {
 	logger                    slog.Logger
 	batchSize                 int32
 	interval                  time.Duration
+	tickTimeout               time.Duration
 	done                      chan struct{}
 }

+// WorkerOption configures a Worker.
+type WorkerOption func(*Worker)
+
+// WithTickTimeout sets the maximum duration for a single tick.
+func WithTickTimeout(d time.Duration) WorkerOption {
+	return func(w *Worker) {
+		if d > 0 {
+			w.tickTimeout = d
+		}
+	}
+}
+
 // NewWorker creates a Worker with default batch size and interval.
 func NewWorker(
 	store Store,
@@ -64,8 +89,9 @@ func NewWorker(
 	publisher PublishDiffStatusChangeFunc,
 	clock quartz.Clock,
 	logger slog.Logger,
+	opts ...WorkerOption,
 ) *Worker {
-	return &Worker{
+	w := &Worker{
 		store:                     store,
 		refresher:                 refresher,
 		publishDiffStatusChangeFn: publisher,
@@ -73,8 +99,13 @@ func NewWorker(
 		logger:                    logger,
 		batchSize:                 defaultBatchSize,
 		interval:                  defaultInterval,
+		tickTimeout:               defaultTickTimeout,
 		done:                      make(chan struct{}),
 	}
+	for _, o := range opts {
+		o(w)
+	}
+	return w
 }

 // Start launches the background loop. It blocks until ctx is
@@ -119,9 +150,10 @@ func chatDiffStatusFromRow(row database.AcquireStaleChatDiffStatusesRow) databas
 }

 func (w *Worker) tick(ctx context.Context) {
-	// Set a context equal to w.interval so that we do not hold up processing due to
-	// random unicorn-related events.
-	ctx, cancel := context.WithTimeout(ctx, w.interval)
+	// Use a dedicated tick timeout that is longer than the
+	// polling interval. This gives concurrent HTTP calls enough
+	// headroom without stalling the next tick excessively.
+	ctx, cancel := context.WithTimeout(ctx, w.tickTimeout)
 	defer cancel()

 	acquiredRows, err := w.store.AcquireStaleChatDiffStatuses(ctx, w.batchSize)
@@ -155,11 +187,18 @@ func (w *Worker) tick(ctx context.Context) {
 			w.logger.Debug(ctx, "refresh chat diff status",
 				slog.F("chat_id", res.Request.Row.ChatID),
 				slog.Error(res.Error))
+			// Apply a longer backoff for rows whose owner has
+			// no linked token — retrying every 2 minutes is
+			// pointless until the user links their account.
+			backoff := DiffStatusTTL
+			if errors.Is(res.Error, ErrNoTokenAvailable) {
+				backoff = NoTokenBackoff
+			}
 			// Back off so the row isn't retried immediately.
 			if err := w.store.BackoffChatDiffStatus(ctx,
 				database.BackoffChatDiffStatusParams{
 					ChatID:  res.Request.Row.ChatID,
-					StaleAt: w.clock.Now().UTC().Add(DiffStatusTTL),
+					StaleAt: w.clock.Now().UTC().Add(backoff),
 				},
 			); err != nil {
 				w.logger.Warn(ctx, "backoff failed chat diff status",
@@ -31,6 +31,7 @@ import (
 type testRefresherCfg struct {
 	resolveBranchPR func(context.Context, string, gitprovider.BranchRef) (*gitprovider.PRRef, error)
 	fetchPRStatus   func(context.Context, string, gitprovider.PRRef) (*gitprovider.PRStatus, error)
+	refresherOpts   []gitsync.RefresherOption
 }

 type testRefresherOpt func(*testRefresherCfg)
@@ -39,6 +40,10 @@ func withResolveBranchPR(f func(context.Context, string, gitprovider.BranchRef)
 	return func(c *testRefresherCfg) { c.resolveBranchPR = f }
 }

+func withRefresherOpts(opts ...gitsync.RefresherOption) testRefresherOpt {
+	return func(c *testRefresherCfg) { c.refresherOpts = opts }
+}
+
 // newTestRefresher creates a Refresher backed by mock
 // provider/token resolvers. The provider recognises any origin,
 // resolves branches to a canned PR, and returns a canned PRStatus.
@@ -84,16 +89,16 @@ func newTestRefresher(t *testing.T, clk quartz.Clock, opts ...testRefresherOpt)
 	}

 	logger := slogtest.Make(t, &slogtest.Options{IgnoreErrors: true})
-	return gitsync.NewRefresher(providers, tokens, logger, clk)
+	return gitsync.NewRefresher(providers, tokens, logger, clk, cfg.refresherOpts...)
 }

-// makeAcquiredRow returns an AcquireStaleChatDiffStatusesRow with
-// a non-empty branch/origin so the Refresher goes through the
+// makeAcquiredRowWithBranch returns an AcquireStaleChatDiffStatusesRow with
+// the given branch and a non-empty origin so the Refresher goes through the
 // branch-resolution path.
-func makeAcquiredRow(chatID, ownerID uuid.UUID) database.AcquireStaleChatDiffStatusesRow {
+func makeAcquiredRowWithBranch(chatID, ownerID uuid.UUID, branch string) database.AcquireStaleChatDiffStatusesRow {
 	return database.AcquireStaleChatDiffStatusesRow{
 		ChatID:          chatID,
-		GitBranch:       "feature",
+		GitBranch:       branch,
 		GitRemoteOrigin: "https://github.com/owner/repo",
 		StaleAt:         time.Now().Add(-time.Minute),
 		OwnerID:         ownerID,
@@ -180,7 +185,7 @@ func TestWorker_LimitsToNRows(t *testing.T) {

 	rows := make([]database.AcquireStaleChatDiffStatusesRow, numRows)
 	for i := range rows {
-		rows[i] = makeAcquiredRow(uuid.New(), ownerID)
+		rows[i] = makeAcquiredRowWithBranch(uuid.New(), ownerID, "feature")
 	}

 	ctrl := gomock.NewController(t)
@@ -233,7 +238,7 @@ func TestWorker_RefresherReturnsNilNil_SkipsUpsert(t *testing.T) {
 	store := dbmock.NewMockStore(ctrl)

 	store.EXPECT().AcquireStaleChatDiffStatuses(gomock.Any(), gomock.Any()).
-		Return([]database.AcquireStaleChatDiffStatusesRow{makeAcquiredRow(chatID, ownerID)}, nil)
+		Return([]database.AcquireStaleChatDiffStatusesRow{makeAcquiredRowWithBranch(chatID, ownerID, "feature")}, nil)

 	mClock := quartz.NewMock(t)
 	logger := slogtest.Make(t, &slogtest.Options{IgnoreErrors: true})
@@ -284,8 +289,8 @@ func TestWorker_RefresherError_BacksOffRow(t *testing.T) {

 	store.EXPECT().AcquireStaleChatDiffStatuses(gomock.Any(), gomock.Any()).
 		Return([]database.AcquireStaleChatDiffStatusesRow{
-			makeAcquiredRow(chat1, ownerID),
-			makeAcquiredRow(chat2, ownerID),
+			makeAcquiredRowWithBranch(chat1, ownerID, "fail-branch"),
+			makeAcquiredRowWithBranch(chat2, ownerID, "success-branch"),
 		}, nil)
 	store.EXPECT().BackoffChatDiffStatus(gomock.Any(), gomock.Any()).
 		DoAndReturn(func(_ context.Context, arg database.BackoffChatDiffStatusParams) error {
@@ -311,13 +316,12 @@ func TestWorker_RefresherError_BacksOffRow(t *testing.T) {

 	logger := slogtest.Make(t, &slogtest.Options{IgnoreErrors: true})

-	// Fail ResolveBranchPullRequest for the first call, succeed
-	// for the second.
-	var callCount atomic.Int32
+	// Fail ResolveBranchPullRequest based on the branch name
+	// so the behavior is deterministic regardless of execution
+	// order.
 	refresher := newTestRefresher(t, mClock, withResolveBranchPR(
-		func(context.Context, string, gitprovider.BranchRef) (*gitprovider.PRRef, error) {
-			n := callCount.Add(1)
-			if n == 1 {
+		func(_ context.Context, _ string, ref gitprovider.BranchRef) (*gitprovider.PRRef, error) {
+			if ref.Branch == "fail-branch" {
 				return nil, fmt.Errorf("simulated provider error")
 			}
 			return &gitprovider.PRRef{Owner: "o", Repo: "r", Number: 1}, nil
@@ -376,8 +380,8 @@ func TestWorker_UpsertError_ContinuesNextRow(t *testing.T) {

 	store.EXPECT().AcquireStaleChatDiffStatuses(gomock.Any(), gomock.Any()).
 		Return([]database.AcquireStaleChatDiffStatusesRow{
-			makeAcquiredRow(chat1, ownerID),
-			makeAcquiredRow(chat2, ownerID),
+			makeAcquiredRowWithBranch(chat1, ownerID, "feature"),
+			makeAcquiredRowWithBranch(chat2, ownerID, "feature"),
 		}, nil)
 	store.EXPECT().UpsertChatDiffStatus(gomock.Any(), gomock.Any()).
 		DoAndReturn(func(_ context.Context, arg database.UpsertChatDiffStatusParams) (database.ChatDiffStatus, error) {
@@ -903,3 +907,57 @@ func TestRefreshChat_UpsertError(t *testing.T) {
 	assert.Nil(t, result)
 	assert.False(t, publishCalled.Load(), "publish should not be called when upsert fails")
 }
+
+func TestWorker_NoTokenBackoff(t *testing.T) {
+	t.Parallel()
+	ctx := testutil.Context(t, testutil.WaitShort)
+
+	chatID := uuid.New()
+	ownerID := uuid.New()
+
+	var mu sync.Mutex
+	var backoffArgs []database.BackoffChatDiffStatusParams
+	tickDone := make(chan struct{})
+
+	mClock := quartz.NewMock(t)
+
+	ctrl := gomock.NewController(t)
+	store := dbmock.NewMockStore(ctrl)
+
+	store.EXPECT().AcquireStaleChatDiffStatuses(gomock.Any(), gomock.Any()).
+		Return([]database.AcquireStaleChatDiffStatusesRow{
+			makeAcquiredRowWithBranch(chatID, ownerID, "feature"),
+		}, nil)
+	store.EXPECT().BackoffChatDiffStatus(gomock.Any(), gomock.Any()).
+		DoAndReturn(func(_ context.Context, arg database.BackoffChatDiffStatusParams) error {
+			mu.Lock()
+			backoffArgs = append(backoffArgs, arg)
+			mu.Unlock()
+			close(tickDone)
+			return nil
+		})
+
+	// Token resolver returns empty token → ErrNoTokenAvailable.
+	// Provider methods should never be called.
+	prov := &mockProvider{}
+	providers := func(string) gitprovider.Provider { return prov }
+	tokens := func(context.Context, uuid.UUID, string) (*string, error) {
+		return ptr.Ref(""), nil
+	}
+
+	logger := slogtest.Make(t, &slogtest.Options{IgnoreErrors: true})
+	refresher := gitsync.NewRefresher(providers, tokens, logger, mClock)
+	worker := gitsync.NewWorker(store, refresher, nil, mClock, logger)
+
+	tickOnce(ctx, t, mClock, worker, tickDone)
+
+	mu.Lock()
+	defer mu.Unlock()
+	require.Len(t, backoffArgs, 1)
+	assert.Equal(t, chatID, backoffArgs[0].ChatID)
+
+	// The backoff should use NoTokenBackoff (10min), not
+	// DiffStatusTTL (2min).
+	expectedStaleAt := mClock.Now().UTC().Add(gitsync.NoTokenBackoff)
+	assert.WithinDuration(t, expectedStaleAt, backoffArgs[0].StaleAt, time.Second)
+}