diff --git a/coderd/x/chatd/chatd_retry_test.go b/coderd/x/chatd/chatd_retry_test.go index 07f8cd5934..24457dabd4 100644 --- a/coderd/x/chatd/chatd_retry_test.go +++ b/coderd/x/chatd/chatd_retry_test.go @@ -3,13 +3,16 @@ package chatd_test import ( "context" "encoding/json" + "fmt" "sync/atomic" "testing" + "time" "github.com/google/uuid" "github.com/prometheus/client_golang/prometheus" "github.com/stretchr/testify/require" + "cdr.dev/slog/v3" "github.com/coder/coder/v2/coderd/database" "github.com/coder/coder/v2/coderd/database/dbgen" "github.com/coder/coder/v2/coderd/database/dbtestutil" @@ -26,6 +29,7 @@ func TestActiveServer_RetryStatePersistedDuringBackoff(t *testing.T) { ctx := testutil.Context(t, testutil.WaitLong) db, ps := dbtestutil.NewDB(t) clock := quartz.NewMock(t).WithLogger(quartz.NoOpLogger) + sink := testutil.NewFakeSink(t) var calls atomic.Int32 openAIURL := chattest.NewOpenAI(t, func(req *chattest.OpenAIRequest) chattest.OpenAIResponse { if !req.Stream { @@ -39,6 +43,7 @@ func TestActiveServer_RetryStatePersistedDuringBackoff(t *testing.T) { user, org, model := seedChatDependenciesWithProvider(t, db, "openai", openAIURL) server := newActiveTestServer(t, db, ps, func(cfg *chatd.Config) { cfg.Clock = clock + cfg.Logger = sink.Logger() }) chat := createChatThroughServer(ctx, t, db, server, org.ID, user.ID, model.ID, "hello") @@ -65,6 +70,12 @@ func TestActiveServer_RetryStatePersistedDuringBackoff(t *testing.T) { latest, err := db.GetChatByID(ctx, chat.ID) require.NoError(t, err) require.False(t, latest.RetryState.Valid) + entries := retryEntriesWithMessage(sink, "chat generation retrying") + require.Len(t, entries, 1) + require.Equal(t, "generate_assistant", retrySinkFieldValue(t, entries[0].Fields, "action")) + require.Equal(t, "openai", retrySinkFieldValue(t, entries[0].Fields, "provider")) + require.Equal(t, "429", retrySinkFieldValue(t, entries[0].Fields, "status_code")) + require.Equal(t, "false", retrySinkFieldValue(t, entries[0].Fields, "chain_broken")) require.Greater(t, latest.RetryStateVersion, withRetry.RetryStateVersion) messages, err := db.GetChatMessagesByChatID(ctx, database.GetChatMessagesByChatIDParams{ChatID: chat.ID}) require.NoError(t, err) @@ -116,13 +127,18 @@ func TestActiveServer_RetryStreamSilenceTimeoutAndClassification(t *testing.T) { }) }) - t.Run("stream silence timeout retry recovers", func(t *testing.T) { + t.Run("silent stream generation retry recovers", func(t *testing.T) { t.Parallel() ctx := testutil.Context(t, testutil.WaitLong) db, ps := dbtestutil.NewDB(t) - clock := quartz.NewMock(t).WithLogger(quartz.NoOpLogger) reg := prometheus.NewRegistry() + clock := quartz.NewMock(t).WithLogger(quartz.NoOpLogger) + streamGuardTrap := clock.Trap().AfterFunc("streamSilenceGuard") + defer streamGuardTrap.Close() + retryTrap := clock.Trap().NewTimer("chatworker", "generation-retry") + defer retryTrap.Close() + sink := testutil.NewFakeSink(t) var calls atomic.Int32 openAIURL := chattest.NewOpenAI(t, func(req *chattest.OpenAIRequest) chattest.OpenAIResponse { if !req.Stream { @@ -137,27 +153,54 @@ func TestActiveServer_RetryStreamSilenceTimeoutAndClassification(t *testing.T) { user, org, model := seedChatDependenciesWithProvider(t, db, "openai", openAIURL) server := newActiveTestServer(t, db, ps, func(cfg *chatd.Config) { cfg.Clock = clock + cfg.Logger = sink.Logger() cfg.PrometheusRegistry = reg + cfg.PendingChatAcquireInterval = 30 * time.Minute + cfg.ChatHeartbeatInterval = 30 * time.Minute }) chat := createChatThroughServer(ctx, t, db, server, org.ID, user.ID, model.ID, "hello") - advanceUntilProviderCall(ctx, clock, &calls, 1) - advanceToNextTimer(ctx, clock) - advanceUntilProviderCall(ctx, clock, &calls, 2) + firstGuard := streamGuardTrap.MustWait(ctx) + firstGuard.MustRelease(ctx) + waitUntilProviderCall(ctx, t, &calls, 1) + advanceMockClockBy(ctx, t, clock, firstGuard.Duration) + retryTimer := retryTrap.MustWait(ctx) + retryTimer.MustRelease(ctx) + advanceMockClockBy(ctx, t, clock, retryTimer.Duration) + secondGuard := streamGuardTrap.MustWait(ctx) + secondGuard.MustRelease(ctx) + waitUntilProviderCall(ctx, t, &calls, 2) waitForChatStatus(ctx, t, db, chat.ID, database.ChatStatusWaiting) require.Equal(t, int32(2), calls.Load()) messages, err := db.GetChatMessagesByChatID(ctx, database.GetChatMessagesByChatIDParams{ChatID: chat.ID}) require.NoError(t, err) requireTextPart(t, messages[len(messages)-1], "recovered") + require.Empty(t, retryEntriesWithMessage(sink, "chatworker task retrying")) + entries := retryEntriesWithMessage(sink, "chat generation retrying") + require.NotEmpty(t, entries) + require.Equal(t, "generate_assistant", retrySinkFieldValue(t, entries[0].Fields, "action")) + require.Equal(t, string(codersdk.ChatErrorKindStreamSilenceTimeout), retrySinkFieldValue(t, entries[0].Fields, "error_kind")) + require.Equal(t, "openai", retrySinkFieldValue(t, entries[0].Fields, "provider")) requireRetryCounter(t, reg, "coderd_chatd_stream_retries_total", 1, map[string]string{ "provider": "openai", - "model": "gpt-4o-mini", + "model": model.Model, "kind": string(codersdk.ChatErrorKindStreamSilenceTimeout), "chain_broken": "false", }) }) } +func retryEntriesWithMessage(sink *testutil.FakeSink, message string) []slog.SinkEntry { + return sink.Entries(func(e slog.SinkEntry) bool { return e.Message == message }) +} + +func retrySinkFieldValue(t *testing.T, fields slog.Map, name string) string { + t.Helper() + value, ok := sinkFieldValue(fields, name) + require.True(t, ok, "missing log field %q", name) + return fmt.Sprint(value) +} + func requireRetryCounter(t *testing.T, reg *prometheus.Registry, name string, wantValue float64, wantLabels map[string]string) { t.Helper() require.True(t, hasRetryCounter(t, reg, name, wantValue, wantLabels), "metric %s not found", name) @@ -210,6 +253,28 @@ func waitForChatRetryState(ctx context.Context, t *testing.T, db database.Store, return chat } +func waitUntilProviderCall(ctx context.Context, t *testing.T, calls *atomic.Int32, want int32) { + t.Helper() + testutil.Eventually(ctx, t, func(context.Context) bool { + return calls.Load() >= want + }, testutil.IntervalFast) +} + +func advanceMockClockBy(ctx context.Context, t *testing.T, clock *quartz.Mock, d time.Duration) { + t.Helper() + for remaining := d; remaining > 0; { + next, ok := clock.Peek() + require.True(t, ok, "no pending clock event while advancing %s", remaining) + if next > remaining { + clock.Advance(remaining).MustWait(ctx) + return + } + _, waiter := clock.AdvanceNext() + waiter.MustWait(ctx) + remaining -= next + } +} + func advanceUntilProviderCall(ctx context.Context, clock *quartz.Mock, calls *atomic.Int32, want int32) { for calls.Load() < want { advanceToNextTimer(ctx, clock) diff --git a/coderd/x/chatd/generation.go b/coderd/x/chatd/generation.go index afca036c09..c3afe31faa 100644 --- a/coderd/x/chatd/generation.go +++ b/coderd/x/chatd/generation.go @@ -360,17 +360,17 @@ func (s *taskStarter) StartGeneration(ctx context.Context, input chatWorkerTaskS Messages: messages, ChainModeDisabled: chainModeDisabled, } - prepared, err := retryGenerationPhase(ctx, s.waitGenerationPhaseBackoff, func() (generationPrepared, error) { + prepared, err := retryGenerationPhase(ctx, s, "prepare", func() (generationPrepared, error) { return s.server.prepareGeneration(ctx, prepareInput) }) if err != nil { - if errors.Is(err, errTaskExpectedExit) { - return errTaskExpectedExit + if errors.Is(err, errTaskExpectedExit) || errors.Is(err, errTaskRetryable) { + return xerrors.Errorf("prepare generation: %w", err) } return s.finishGenerationError(ctx, machine, input, 0, err, generationAttemptNotRequired) } cleanup := prepared.Cleanup - decision, err := retryGenerationPhase(ctx, s.waitGenerationPhaseBackoff, func() (generationDecision, error) { + decision, err := retryGenerationPhase(ctx, s, "decide", func() (generationDecision, error) { return decideGenerationAction(generationDecisionInput{ chat: prepared.Chat, messages: prepared.Messages, @@ -387,8 +387,8 @@ func (s *taskStarter) StartGeneration(ctx context.Context, input chatWorkerTaskS }) if err != nil { cleanup() - if errors.Is(err, errTaskExpectedExit) { - return errTaskExpectedExit + if errors.Is(err, errTaskExpectedExit) || errors.Is(err, errTaskRetryable) { + return xerrors.Errorf("decide generation: %w", err) } if errors.Is(err, errCompactionStillOverLimit) && prepared.Compaction != nil { s.server.metrics.RecordCompaction( @@ -424,19 +424,33 @@ func (s *taskStarter) StartGeneration(ctx context.Context, input chatWorkerTaskS if actionErr == nil { return nil } - if errors.Is(actionErr, errTaskExpectedExit) || errors.Is(actionErr, chatloop.ErrInterrupted) { - return nil + // Task cancellation is handled by the runner, not here. + if ctx.Err() != nil && errors.Is(actionErr, context.Canceled) { + return errors.Join(errTaskExpectedExit, xerrors.Errorf("generation action: %w", actionErr), ctx.Err()) } - if errors.Is(actionErr, context.Canceled) && ctx.Err() != nil { + if errors.Is(actionErr, errTaskExpectedExit) || errors.Is(actionErr, chatloop.ErrInterrupted) { return nil } classified := chaterror.Classify(actionErr) if classified.Retryable { + action := decision.kind decision, err := s.recordGenerationRetry(ctx, machine, input, classified) if err != nil { return err } if decision.retry { + s.opts.Logger.Warn(ctx, "chat generation retrying", + slog.F("chat_id", input.ChatID), + slog.F("worker_id", input.WorkerID), + slog.F("action", action), + slog.F("generation_attempt", decision.generationAttempt), + slog.F("delay", decision.delay), + slog.F("error_kind", classified.Kind), + slog.F("provider", classified.Provider), + slog.F("status_code", classified.StatusCode), + slog.F("chain_broken", classified.ChainBroken), + slogError(actionErr), + ) if classified.ChainBroken { chainModeDisabled = true } @@ -548,7 +562,7 @@ func (s *taskStarter) waitGenerationRetry(ctx context.Context, delay time.Durati case <-timer.C: return nil case <-ctx.Done(): - return errTaskExpectedExit + return errors.Join(errTaskExpectedExit, xerrors.Errorf("wait generation retry: %w", ctx.Err())) } } @@ -576,12 +590,9 @@ func generationPhaseBackoff(attempt int) time.Duration { // returns early on success or on a terminal error (see terminalGeneration). // Non-terminal errors are retried with exponential backoff. Context // cancellation returns errTaskExpectedExit so shutdown does not write an -// error state. When every attempt fails, the last error is returned. -func retryGenerationPhase[T any]( - ctx context.Context, - wait func(context.Context, time.Duration) error, - fn func() (T, error), -) (T, error) { +// error state. Task timeouts return a retryable task error so the runner can +// start a fresh attempt. When every attempt fails, the last error is returned. +func retryGenerationPhase[T any](ctx context.Context, starter *taskStarter, phase string, fn func() (T, error)) (T, error) { var zero T var lastErr error for attempt := 0; attempt < generationPhaseMaxAttempts; attempt++ { @@ -590,14 +601,22 @@ func retryGenerationPhase[T any]( return result, nil } if isTerminalGeneration(err) { - return zero, err + return zero, xerrors.Errorf("retryGenerationPhase terminal error: %w", err) } if ctx.Err() != nil { - return zero, errTaskExpectedExit + return zero, errors.Join(errTaskExpectedExit, xerrors.Errorf("retryGenerationPhase %s: %w", phase, ctx.Err())) } lastErr = err if attempt < generationPhaseMaxAttempts-1 { - if waitErr := wait(ctx, generationPhaseBackoff(attempt)); waitErr != nil { + delay := generationPhaseBackoff(attempt) + starter.opts.Logger.Warn(ctx, "chat generation phase retrying", + slog.F("phase", phase), + slog.F("attempt", attempt+1), + slog.F("max_attempts", generationPhaseMaxAttempts), + slog.F("delay", delay), + slogError(err), + ) + if waitErr := starter.waitGenerationPhaseBackoff(ctx, delay); waitErr != nil { return zero, waitErr } } @@ -612,7 +631,7 @@ func (s *taskStarter) waitGenerationPhaseBackoff(ctx context.Context, delay time case <-timer.C: return nil case <-ctx.Done(): - return errTaskExpectedExit + return errors.Join(errTaskExpectedExit, xerrors.Errorf("wait generation phase backoff: %w", ctx.Err())) } } diff --git a/coderd/x/chatd/generation_retry_internal_test.go b/coderd/x/chatd/generation_retry_internal_test.go index 87ef475ca5..f55b350295 100644 --- a/coderd/x/chatd/generation_retry_internal_test.go +++ b/coderd/x/chatd/generation_retry_internal_test.go @@ -7,6 +7,9 @@ import ( "github.com/stretchr/testify/require" "golang.org/x/xerrors" + + "github.com/coder/coder/v2/testutil" + "github.com/coder/quartz" ) func TestTerminalGeneration(t *testing.T) { @@ -36,92 +39,100 @@ func TestGenerationPhaseBackoff(t *testing.T) { func TestRetryGenerationPhase(t *testing.T) { t.Parallel() - noopWait := func(context.Context, time.Duration) error { return nil } - t.Run("SuccessFirstTry", func(t *testing.T) { t.Parallel() + starter := newGenerationPhaseTestStarter(t, quartz.NewMock(t).WithLogger(quartz.NoOpLogger)) calls := 0 - waits := 0 - wait := func(context.Context, time.Duration) error { - waits++ - return nil - } - got, err := retryGenerationPhase(context.Background(), wait, func() (int, error) { + got, err := retryGenerationPhase(context.Background(), starter, "prepare", func() (int, error) { calls++ return 42, nil }) require.NoError(t, err) require.Equal(t, 42, got) require.Equal(t, 1, calls) - require.Equal(t, 0, waits) }) t.Run("RetryThenSuccess", func(t *testing.T) { t.Parallel() + clock := quartz.NewMock(t).WithLogger(quartz.NoOpLogger) + timerTrap := clock.Trap().NewTimer("chatworker", "generation-phase-retry") + defer timerTrap.Close() + sink := testutil.NewFakeSink(t) + starter := newGenerationPhaseTestStarter(t, clock) + starter.opts.Logger = sink.Logger() + ctx := testutil.Context(t, testutil.WaitLong) calls := 0 - waits := 0 - var delays []time.Duration - wait := func(_ context.Context, d time.Duration) error { - waits++ - delays = append(delays, d) - return nil - } - got, err := retryGenerationPhase(context.Background(), wait, func() (string, error) { - calls++ - if calls < 2 { - return "", xerrors.New("transient") - } - return "ok", nil - }) - require.NoError(t, err) - require.Equal(t, "ok", got) + done := make(chan phaseRetryResult[string], 1) + go func() { + got, err := retryGenerationPhase(ctx, starter, "prepare", func() (string, error) { + calls++ + if calls < 2 { + return "", xerrors.New("transient") + } + return "ok", nil + }) + done <- phaseRetryResult[string]{value: got, err: err} + }() + + timerTrap.MustWait(ctx).MustRelease(ctx) + clock.Advance(generationPhaseBackoff(0)).MustWait(ctx) + result := <-done + require.NoError(t, result.err) + require.Equal(t, "ok", result.value) require.Equal(t, 2, calls) - require.Equal(t, 1, waits) - require.Equal(t, []time.Duration{generationPhaseBackoff(0)}, delays) + entries := entriesWithMessage(sink, "chat generation phase retrying") + require.Len(t, entries, 1) + require.Equal(t, "prepare", sinkFieldValue(t, entries[0].Fields, "phase")) + require.Equal(t, "1", sinkFieldValue(t, entries[0].Fields, "attempt")) + require.Equal(t, generationPhaseBackoff(0).String(), sinkFieldValue(t, entries[0].Fields, "delay")) }) t.Run("ExhaustsAndReturnsLastError", func(t *testing.T) { t.Parallel() + clock := quartz.NewMock(t).WithLogger(quartz.NoOpLogger) + timerTrap := clock.Trap().NewTimer("chatworker", "generation-phase-retry") + defer timerTrap.Close() + starter := newGenerationPhaseTestStarter(t, clock) + ctx := testutil.Context(t, testutil.WaitLong) calls := 0 - waits := 0 - wait := func(context.Context, time.Duration) error { - waits++ - return nil - } - _, err := retryGenerationPhase(context.Background(), wait, func() (int, error) { - calls++ - return 0, xerrors.Errorf("attempt %d", calls) - }) - require.EqualError(t, err, "attempt 3") + done := make(chan error, 1) + go func() { + _, err := retryGenerationPhase(ctx, starter, "prepare", func() (int, error) { + calls++ + return 0, xerrors.Errorf("attempt %d", calls) + }) + done <- err + }() + + timerTrap.MustWait(ctx).MustRelease(ctx) + clock.Advance(generationPhaseBackoff(0)).MustWait(ctx) + timerTrap.MustWait(ctx).MustRelease(ctx) + clock.Advance(generationPhaseBackoff(1)).MustWait(ctx) + require.EqualError(t, <-done, "attempt 3") require.Equal(t, generationPhaseMaxAttempts, calls) - require.Equal(t, generationPhaseMaxAttempts-1, waits) }) t.Run("TerminalShortCircuits", func(t *testing.T) { t.Parallel() + starter := newGenerationPhaseTestStarter(t, quartz.NewMock(t).WithLogger(quartz.NoOpLogger)) calls := 0 - waits := 0 - wait := func(context.Context, time.Duration) error { - waits++ - return nil - } cause := xerrors.New("deterministic") - _, err := retryGenerationPhase(context.Background(), wait, func() (int, error) { + _, err := retryGenerationPhase(context.Background(), starter, "prepare", func() (int, error) { calls++ return 0, terminalGeneration(cause) }) require.ErrorIs(t, err, cause) require.True(t, isTerminalGeneration(err)) require.Equal(t, 1, calls) - require.Equal(t, 0, waits) }) t.Run("ContextCanceledExitsCleanly", func(t *testing.T) { t.Parallel() + starter := newGenerationPhaseTestStarter(t, quartz.NewMock(t).WithLogger(quartz.NoOpLogger)) ctx, cancel := context.WithCancel(context.Background()) cancel() calls := 0 - _, err := retryGenerationPhase(ctx, noopWait, func() (int, error) { + _, err := retryGenerationPhase(ctx, starter, "prepare", func() (int, error) { calls++ return 0, xerrors.New("transient") }) @@ -131,18 +142,40 @@ func TestRetryGenerationPhase(t *testing.T) { t.Run("WaitCancellationExitsCleanly", func(t *testing.T) { t.Parallel() + clock := quartz.NewMock(t).WithLogger(quartz.NoOpLogger) + timerTrap := clock.Trap().NewTimer("chatworker", "generation-phase-retry") + defer timerTrap.Close() + starter := newGenerationPhaseTestStarter(t, clock) + ctx, cancel := context.WithCancel(testutil.Context(t, testutil.WaitLong)) calls := 0 - waits := 0 - wait := func(context.Context, time.Duration) error { - waits++ - return errTaskExpectedExit - } - _, err := retryGenerationPhase(context.Background(), wait, func() (int, error) { - calls++ - return 0, xerrors.New("transient") - }) - require.ErrorIs(t, err, errTaskExpectedExit) + done := make(chan error, 1) + go func() { + _, err := retryGenerationPhase(ctx, starter, "prepare", func() (int, error) { + calls++ + return 0, xerrors.New("transient") + }) + done <- err + }() + + timerTrap.MustWait(ctx).MustRelease(ctx) + cancel() + require.ErrorIs(t, <-done, errTaskExpectedExit) require.Equal(t, 1, calls) - require.Equal(t, 1, waits) }) } + +type phaseRetryResult[T any] struct { + value T + err error +} + +func newGenerationPhaseTestStarter(t *testing.T, clock quartz.Clock) *taskStarter { + t.Helper() + require.NotNil(t, clock) + return &taskStarter{opts: chatWorkerOptions{ + Clock: clock, + Logger: testutil.NewFakeSink(t).Logger(), + TaskRetryInitialBackoff: time.Millisecond, + TaskRetryMaxBackoff: time.Millisecond, + }} +} diff --git a/coderd/x/chatd/runner.go b/coderd/x/chatd/runner.go index 5498d4d769..41109f7a81 100644 --- a/coderd/x/chatd/runner.go +++ b/coderd/x/chatd/runner.go @@ -232,11 +232,11 @@ func (r *runner) runTask( err := runTaskWithRetry(ctx, r.opts.retryOptions(), kind, func(ctx context.Context) error { unlock, ok := r.localLocks.acquire(ctx, key) if !ok { - return errTaskExpectedExit + return errors.Join(errTaskExpectedExit, xerrors.Errorf("runTask acquire local lock: %w", ctx.Err())) } defer unlock() if ctx.Err() != nil { - return errTaskExpectedExit + return errors.Join(errTaskExpectedExit, xerrors.Errorf("runTask context canceled: %w", ctx.Err())) } switch kind { diff --git a/coderd/x/chatd/runner_test.go b/coderd/x/chatd/runner_test.go index eca1df9a26..ef95069518 100644 --- a/coderd/x/chatd/runner_test.go +++ b/coderd/x/chatd/runner_test.go @@ -1,6 +1,7 @@ package chatd //nolint:testpackage // Uses unexported chatworker helpers. import ( + "context" "testing" "time" @@ -38,6 +39,7 @@ func TestRunner_CancelsActiveTaskWhenHistoryChanges(t *testing.T) { updated := commitAssistantStep(t, f, chat.ID, "first step") require.Greater(t, updated.HistoryVersion, first.input.HistoryVersion) requireTaskCanceled(t, first) + require.NotErrorIs(t, context.Cause(first.ctx), errTaskTimeout) second := starter.waitCall(t, taskKindGeneration, chat.ID) require.Equal(t, updated.HistoryVersion, second.input.HistoryVersion) } @@ -104,11 +106,39 @@ func TestRunner_AllowsReplacementForDifferentHistoryOrStatus(t *testing.T) { require.Equal(t, updated.HistoryVersion, second.input.HistoryVersion) } +func TestRunner_TaskTimeoutRetries(t *testing.T) { + t.Parallel() + f := newWorkerTestFixture(t) + chat := f.createRunningChat(t) + clock := quartz.NewMock(t).WithLogger(quartz.NoOpLogger) + timeoutTrap := clock.Trap().AfterFunc("chatworker", "task-timeout-generation") + starter := newBlockingTaskStarter(false) + opts := testOptions(t, f, starter) + opts.Clock = clock + opts.TaskRetryInitialBackoff = time.Minute + opts.TaskRetryMaxBackoff = time.Minute + startWorker(t, opts) + + timeoutTrap.MustWait(testutil.Context(t, testutil.WaitLong)).MustRelease(testutil.Context(t, testutil.WaitLong)) + timeoutTrap.Close() + first := starter.waitCall(t, taskKindGeneration, chat.ID) + retryTrap := clock.Trap().NewTimer("chatworker", "task-retry-generation") + defer retryTrap.Close() + + ctx := testutil.Context(t, testutil.WaitLong) + clock.Advance(defaultTaskTimeout).MustWait(ctx) + retryTrap.MustWait(ctx).MustRelease(ctx) + require.ErrorIs(t, context.Cause(first.ctx), errTaskTimeout) + clock.Advance(time.Minute).MustWait(ctx) + second := starter.waitCall(t, taskKindGeneration, chat.ID) + require.Equal(t, first.input.HistoryVersion, second.input.HistoryVersion) +} + func TestWorker_RoutesDatabaseSyncStateToActiveRunner(t *testing.T) { t.Parallel() f := newWorkerTestFixture(t) chat := f.createRunningChat(t) - clock := quartz.NewMock(t) + clock := quartz.NewMock(t).WithLogger(quartz.NoOpLogger) starter := newBlockingTaskStarter(false) opts := testOptions(t, f, starter) opts.Clock = clock diff --git a/coderd/x/chatd/tasks.go b/coderd/x/chatd/tasks.go index 8b2be67c7a..f370ea852f 100644 --- a/coderd/x/chatd/tasks.go +++ b/coderd/x/chatd/tasks.go @@ -11,6 +11,7 @@ import ( "github.com/google/uuid" "golang.org/x/xerrors" + "cdr.dev/slog/v3" "github.com/coder/coder/v2/coderd/database" coderdpubsub "github.com/coder/coder/v2/coderd/pubsub" "github.com/coder/coder/v2/coderd/x/chatd/chatdebug" @@ -21,11 +22,18 @@ import ( "github.com/coder/quartz" ) -const postCommitWatchPublishTimeout = 10 * time.Second +const ( + postCommitWatchPublishTimeout = 10 * time.Second + // defaultTaskTimeout must exceed chatloop's stream-silence guard so + // silent provider streams fail through chat-specific retry handling + // before the runner retries the whole task. + defaultTaskTimeout = 15 * time.Minute +) var ( errTaskExpectedExit = xerrors.New("chatworker task expected exit") errTaskRetryable = xerrors.New("chatworker task retryable error") + errTaskTimeout = xerrors.New("chatworker task timeout") ) type taskRetryableError struct { @@ -48,10 +56,15 @@ func (e taskRetryableError) Unwrap() error { type retryWrapperOptions struct { clock quartz.Clock + logger slog.Logger initialDelay time.Duration maxDelay time.Duration } +// runTaskWithRetry ensures that a task doesn't exit until it completes +// successfully or gets canceled. It retries the task in case of any ephemeral errors. +// It's critical for the correct operation of the chat runner: +// this function is THE place that ensures task liveness within the runner. func runTaskWithRetry( ctx context.Context, opts retryWrapperOptions, @@ -70,19 +83,46 @@ func runTaskWithRetry( if opts.maxDelay < opts.initialDelay { opts.maxDelay = opts.initialDelay } - delay := opts.initialDelay for { - err := executeTaskSafely(ctx, fn) - switch { - case err == nil: - return nil - case errors.Is(err, errTaskExpectedExit): - return nil - case ctx.Err() != nil: + attemptCtx, cancelAttempt := taskAttemptContext(ctx, opts.clock, kind) + err := executeTaskSafely(attemptCtx, fn) + timedOut := errors.Is(context.Cause(attemptCtx), errTaskTimeout) + cancelAttempt() + if timedOut && err != nil { + if !errors.Is(err, errTaskExpectedExit) || + errors.Is(err, context.Canceled) || + errors.Is(err, context.DeadlineExceeded) || + errors.Is(err, errTaskTimeout) { + err = taskRetryableError{err: errors.Join(errTaskTimeout, err)} + } + } + if err == nil { + // no log on success to avoid noise return nil } + exitReason := "" + switch { + case ctx.Err() != nil: + exitReason = "context_canceled" + case errors.Is(err, errTaskExpectedExit) && !errors.Is(err, errTaskRetryable): + exitReason = "expected_non_retryable_exit" + } + if exitReason != "" { + opts.logger.Debug(ctx, "chatworker task exited", + slog.F("task_kind", kind), + slog.F("reason", exitReason), + slogError(err), + ) + return nil + } + + opts.logger.Warn(ctx, "chatworker task retrying", + slog.F("task_kind", kind), + slog.F("delay", delay), + slogError(err), + ) timer := opts.clock.NewTimer(delay, "chatworker", "task-retry-"+string(kind)) select { case <-timer.C: @@ -100,6 +140,17 @@ func runTaskWithRetry( } } +func taskAttemptContext(ctx context.Context, clock quartz.Clock, kind taskKind) (context.Context, func()) { + attemptCtx, cancelCause := context.WithCancelCause(ctx) + timer := clock.AfterFunc(defaultTaskTimeout, func() { + cancelCause(errTaskTimeout) + }, "chatworker", "task-timeout-"+string(kind)) + return attemptCtx, func() { + timer.Stop() + cancelCause(nil) + } +} + func executeTaskSafely(ctx context.Context, fn func(context.Context) error) (err error) { defer func() { if recovered := recover(); recovered != nil { @@ -167,6 +218,7 @@ func newTaskStarter( func (o chatWorkerOptions) retryOptions() retryWrapperOptions { return retryWrapperOptions{ clock: o.Clock, + logger: o.Logger, initialDelay: o.TaskRetryInitialBackoff, maxDelay: o.TaskRetryMaxBackoff, } @@ -200,7 +252,7 @@ func (s *taskStarter) StartInterrupt(ctx context.Context, input chatWorkerTaskSt } if err := s.opts.MessagePartBuffer.CloseEpisode(key); err != nil { if ctx.Err() != nil { - return errTaskExpectedExit + return errors.Join(errTaskExpectedExit, xerrors.Errorf("close message part episode: %w", err), ctx.Err()) } return taskRetryableError{err: xerrors.Errorf("close message part episode: %w", err)} } @@ -211,7 +263,7 @@ func (s *taskStarter) StartInterrupt(ctx context.Context, input chatWorkerTaskSt } if err != nil { if ctx.Err() != nil { - return errTaskExpectedExit + return errors.Join(errTaskExpectedExit, xerrors.Errorf("get message part episode: %w", err), ctx.Err()) } return taskRetryableError{err: xerrors.Errorf("get message part episode: %w", err)} } @@ -362,7 +414,7 @@ func (s *taskStarter) waitUntil(ctx context.Context, deadline time.Time) error { case <-timer.C: return nil case <-ctx.Done(): - return errTaskExpectedExit + return errors.Join(errTaskExpectedExit, xerrors.Errorf("wait until: %w", ctx.Err())) } } @@ -491,14 +543,14 @@ func (s *taskStarter) publishWatchWithRetry( if err := publishChatWatchEvent(s.opts.Pubsub, chat, kind); err == nil { return nil } else if ctx.Err() != nil { - return errTaskExpectedExit + return errors.Join(errTaskExpectedExit, xerrors.Errorf("publishChatWatchEvent: %w", ctx.Err())) } timer := s.opts.Clock.NewTimer(delay, "chatworker", "watch-publish-retry") select { case <-timer.C: case <-ctx.Done(): timer.Stop() - return errTaskExpectedExit + return errors.Join(errTaskExpectedExit, xerrors.Errorf("watch publish retry context done: %w", ctx.Err())) } timer.Stop() if delay < s.opts.TaskRetryMaxBackoff { @@ -560,8 +612,11 @@ func normalizeTaskInfrastructureError(err error, action string) error { if err == nil { return nil } - if errors.Is(err, errTaskExpectedExit) || errors.Is(err, chatstate.ErrChatNotFound) || errors.Is(err, sql.ErrNoRows) || errors.Is(err, context.Canceled) { - return errTaskExpectedExit + if errors.Is(err, errTaskExpectedExit) { + return err + } + if errors.Is(err, chatstate.ErrChatNotFound) || errors.Is(err, sql.ErrNoRows) || errors.Is(err, context.Canceled) { + return errors.Join(errTaskExpectedExit, xerrors.Errorf("%s: %w", action, err)) } return taskRetryableError{err: xerrors.Errorf("%s: %w", action, err)} } @@ -570,8 +625,11 @@ func normalizeTaskTransitionError(err error, action string) error { if err == nil { return nil } - if errors.Is(err, errTaskExpectedExit) || errors.Is(err, chatstate.ErrChatNotFound) || errors.Is(err, sql.ErrNoRows) || errors.Is(err, context.Canceled) { - return errTaskExpectedExit + if errors.Is(err, errTaskExpectedExit) { + return err + } + if errors.Is(err, chatstate.ErrChatNotFound) || errors.Is(err, sql.ErrNoRows) || errors.Is(err, context.Canceled) { + return errors.Join(errTaskExpectedExit, xerrors.Errorf("%s: %w", action, err)) } if errors.Is(err, chatstate.ErrTransitionNotAllowed) || errors.Is(err, chatstate.ErrInvalidState) { return xerrors.Errorf("%s: %w", action, err) diff --git a/coderd/x/chatd/tasks_test.go b/coderd/x/chatd/tasks_test.go index a4540a1537..8fcc588c35 100644 --- a/coderd/x/chatd/tasks_test.go +++ b/coderd/x/chatd/tasks_test.go @@ -5,6 +5,7 @@ import ( "context" "database/sql" "encoding/json" + "fmt" "sync" "testing" "time" @@ -34,9 +35,11 @@ func TestRetryWrapper_ExpectedExitsDoNotRetry(t *testing.T) { t.Parallel() ctx := testutil.Context(t, testutil.WaitShort) + sink := testutil.NewFakeSink(t) calls := 0 err := runTaskWithRetry(ctx, retryWrapperOptions{ clock: quartz.NewMock(t), + logger: sink.Logger(), initialDelay: time.Second, maxDelay: time.Second, }, taskKindInterrupt, func(context.Context) error { @@ -45,6 +48,7 @@ func TestRetryWrapper_ExpectedExitsDoNotRetry(t *testing.T) { }) require.NoError(t, err) require.Equal(t, 1, calls) + require.Empty(t, entriesWithMessage(sink, "chatworker task retrying")) } func TestRetryWrapper_UnexpectedErrorsRetry(t *testing.T) { @@ -54,11 +58,13 @@ func TestRetryWrapper_UnexpectedErrorsRetry(t *testing.T) { trap := clock.Trap().NewTimer("chatworker", "task-retry-requires_action_timeout") defer trap.Close() ctx := testutil.Context(t, testutil.WaitLong) + sink := testutil.NewFakeSink(t) calls := 0 done := make(chan error, 1) go func() { done <- runTaskWithRetry(ctx, retryWrapperOptions{ clock: clock, + logger: sink.Logger(), initialDelay: time.Minute, maxDelay: time.Minute, }, taskKindRequiresActionTimeout, func(context.Context) error { @@ -74,6 +80,11 @@ func TestRetryWrapper_UnexpectedErrorsRetry(t *testing.T) { clock.Advance(time.Minute).MustWait(ctx) require.NoError(t, <-done) require.Equal(t, 2, calls) + entries := entriesWithMessage(sink, "chatworker task retrying") + require.Len(t, entries, 1) + require.Equal(t, string(taskKindRequiresActionTimeout), sinkFieldValue(t, entries[0].Fields, "task_kind")) + require.Equal(t, time.Minute.String(), sinkFieldValue(t, entries[0].Fields, "delay")) + require.Contains(t, sinkFieldValue(t, entries[0].Fields, "error"), "database unavailable") } func TestRetryWrapper_PanicsRetry(t *testing.T) { @@ -83,11 +94,13 @@ func TestRetryWrapper_PanicsRetry(t *testing.T) { trap := clock.Trap().NewTimer("chatworker", "task-retry-generation") defer trap.Close() ctx := testutil.Context(t, testutil.WaitLong) + sink := testutil.NewFakeSink(t) calls := 0 done := make(chan error, 1) go func() { done <- runTaskWithRetry(ctx, retryWrapperOptions{ clock: clock, + logger: sink.Logger(), initialDelay: time.Minute, maxDelay: time.Minute, }, taskKindGeneration, func(context.Context) error { @@ -103,6 +116,118 @@ func TestRetryWrapper_PanicsRetry(t *testing.T) { clock.Advance(time.Minute).MustWait(ctx) require.NoError(t, <-done) require.Equal(t, 2, calls) + entries := entriesWithMessage(sink, "chatworker task retrying") + require.Len(t, entries, 1) + require.Contains(t, sinkFieldValue(t, entries[0].Fields, "error"), "chatworker task panic: database unavailable") +} + +// database/sql returns ctx.Err() from ctxDriverQuery, not +// context.Cause(ctx). This test checks that the retry logic +// doesn't classify such an error as an expected exit when +// task timeout is the cause of the cancellation. +func TestRetryWrapper_TaskTimeoutDBQueryCancellationRetries(t *testing.T) { + t.Parallel() + + f := newTaskTestFixture(t) + clock := quartz.NewMock(t) + timeoutTrap := clock.Trap().AfterFunc("chatworker", "task-timeout-generation") + retryTrap := clock.Trap().NewTimer("chatworker", "task-retry-generation") + defer retryTrap.Close() + ctx := testutil.Context(t, testutil.WaitLong) + sink := testutil.NewFakeSink(t) + calls := 0 + firstCallStarted := make(chan struct{}) + var firstQueryErr error + var firstQueryCause error + done := make(chan error, 1) + go func() { + done <- runTaskWithRetry(ctx, retryWrapperOptions{ + clock: clock, + logger: sink.Logger(), + initialDelay: time.Minute, + maxDelay: time.Minute, + }, taskKindGeneration, func(ctx context.Context) error { + calls++ + if calls == 1 { + close(firstCallStarted) + <-ctx.Done() + _, err := f.db.GetDatabaseNow(ctx) + firstQueryErr = err + firstQueryCause = context.Cause(ctx) + return normalizeTaskTransitionError(err, "db query") + } + return nil + }) + }() + + timeoutTrap.MustWait(ctx).MustRelease(ctx) + timeoutTrap.Close() + <-firstCallStarted + clock.Advance(defaultTaskTimeout).MustWait(ctx) + retryTrap.MustWait(ctx).MustRelease(ctx) + clock.Advance(time.Minute).MustWait(ctx) + require.NoError(t, <-done) + require.Equal(t, 2, calls) + require.ErrorIs(t, firstQueryErr, context.Canceled) + require.NotErrorIs(t, firstQueryErr, errTaskTimeout) + require.ErrorIs(t, firstQueryCause, errTaskTimeout) + entries := entriesWithMessage(sink, "chatworker task retrying") + require.Len(t, entries, 1) + require.Contains(t, sinkFieldValue(t, entries[0].Fields, "error"), errTaskTimeout.Error()) + require.Contains(t, sinkFieldValue(t, entries[0].Fields, "error"), context.Canceled.Error()) +} + +func TestRetryWrapper_ContextCancellationDoesNotRetryOrLog(t *testing.T) { + t.Parallel() + + ctx, cancel := context.WithCancel(testutil.Context(t, testutil.WaitLong)) + cancel() + sink := testutil.NewFakeSink(t) + calls := 0 + original := xerrors.New("database unavailable") + err := runTaskWithRetry(ctx, retryWrapperOptions{ + clock: quartz.NewMock(t), + logger: sink.Logger(), + initialDelay: time.Second, + maxDelay: time.Second, + }, taskKindGeneration, func(context.Context) error { + calls++ + return original + }) + require.NoError(t, err) + require.Equal(t, 1, calls) + require.Empty(t, entriesWithMessage(sink, "chatworker task retrying")) +} + +func TestNormalizeTaskErrors_ContextCancellationIsExpectedExit(t *testing.T) { + t.Parallel() + + err := normalizeTaskInfrastructureError(context.Canceled, "lock chat") + require.ErrorIs(t, err, errTaskExpectedExit) + require.ErrorIs(t, err, context.Canceled) + require.NotErrorIs(t, err, errTaskRetryable) + require.NotErrorIs(t, err, errTaskTimeout) + + err = normalizeTaskTransitionError(context.Canceled, "commit chat") + require.ErrorIs(t, err, errTaskExpectedExit) + require.ErrorIs(t, err, context.Canceled) + require.NotErrorIs(t, err, errTaskRetryable) + require.NotErrorIs(t, err, errTaskTimeout) +} + +func entriesWithMessage(sink *testutil.FakeSink, message string) []slog.SinkEntry { + return sink.Entries(func(e slog.SinkEntry) bool { return e.Message == message }) +} + +func sinkFieldValue(t *testing.T, fields slog.Map, name string) string { + t.Helper() + for _, f := range fields { + if f.Name == name { + return fmt.Sprint(f.Value) + } + } + t.Fatalf("missing log field %q", name) + return "" } func TestInterruptTask_FinishInterruptionOnly(t *testing.T) {