From b1e80e6f3aa4131c6819c7ec5523e2ad823a9992 Mon Sep 17 00:00:00 2001 From: Kyle Carberry Date: Thu, 12 Mar 2026 11:08:06 -0700 Subject: [PATCH] fix(gitsync): concurrent refresh, decoupled timeout, and no-token backoff (#23004) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit ## 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`. --- coderd/gitsync/gitsync.go | 150 +++++++++++++++++++++++++------ coderd/gitsync/gitsync_test.go | 156 +++++++++++++++++++++------------ coderd/gitsync/worker.go | 49 +++++++++-- coderd/gitsync/worker_test.go | 92 +++++++++++++++---- 4 files changed, 342 insertions(+), 105 deletions(-) diff --git a/coderd/gitsync/gitsync.go b/coderd/gitsync/gitsync.go index 0ac842b0ec..6f2507b612 100644 --- a/coderd/gitsync/gitsync.go +++ b/coderd/gitsync/gitsync.go @@ -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, diff --git a/coderd/gitsync/gitsync_test.go b/coderd/gitsync/gitsync_test.go index 6d720131ff..1f81f78616 100644 --- a/coderd/gitsync/gitsync_test.go +++ b/coderd/gitsync/gitsync_test.go @@ -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()) +} diff --git a/coderd/gitsync/worker.go b/coderd/gitsync/worker.go index 17992ba95d..8f99c884c2 100644 --- a/coderd/gitsync/worker.go +++ b/coderd/gitsync/worker.go @@ -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", diff --git a/coderd/gitsync/worker_test.go b/coderd/gitsync/worker_test.go index 8a228bd614..7d27f7a992 100644 --- a/coderd/gitsync/worker_test.go +++ b/coderd/gitsync/worker_test.go @@ -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) +}