From c0c0d1e35338f804f63c5eac147cfd83d9eb8f3a Mon Sep 17 00:00:00 2001 From: Hugo Dutka Date: Thu, 18 Jun 2026 19:02:42 +0200 Subject: [PATCH] chore(coderd/x/chatd): improve runner logging (#26522) Follow up to https://github.com/coder/coder/pull/26412. - Ensure that each `errTaskExpectedExit` and any other error is joined with a descriptive reason. This will enable [`runChatWithRetry`](https://github.com/coder/coder/blob/5601ea18ed650deafa4a2da7286a144f9f382330/coderd/x/chatd/tasks.go#L113) to log a descriptive message when a task exits. - Pass additional task information like chat id into `runTaskWithRetry` so logs can be more descriptive. --- coderd/x/chatd/generation.go | 94 +++++++++++++++++++----------------- coderd/x/chatd/runner.go | 7 ++- coderd/x/chatd/tasks.go | 63 ++++++++++++++---------- coderd/x/chatd/tasks_test.go | 10 ++-- 4 files changed, 99 insertions(+), 75 deletions(-) diff --git a/coderd/x/chatd/generation.go b/coderd/x/chatd/generation.go index 103ddb301b..de8a42d2d6 100644 --- a/coderd/x/chatd/generation.go +++ b/coderd/x/chatd/generation.go @@ -353,7 +353,7 @@ func (s *taskStarter) StartGeneration(ctx context.Context, input chatWorkerTaskS for { locked, messages, err := loadGenerationState(ctx, machine, input) if err != nil { - return err + return xerrors.Errorf("load generation state: %w", err) } prepareInput := generationPrepareInput{ Chat: locked, @@ -428,15 +428,18 @@ func (s *taskStarter) StartGeneration(ctx context.Context, input chatWorkerTaskS 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, errTaskExpectedExit) || errors.Is(actionErr, chatloop.ErrInterrupted) { - return nil + if errors.Is(actionErr, chatloop.ErrInterrupted) { + return errors.Join(errTaskExpectedExit, xerrors.Errorf("generation action: %w", actionErr)) + } + if errors.Is(actionErr, errTaskExpectedExit) { + return xerrors.Errorf("generation action: %w", actionErr) } classified := chaterror.Classify(actionErr) if classified.Retryable { action := decision.kind decision, err := s.recordGenerationRetry(ctx, machine, input, classified) if err != nil { - return err + return xerrors.Errorf("record generation retry: %w", err) } if decision.retry { s.opts.Logger.Warn(ctx, "chat generation retrying", @@ -455,7 +458,7 @@ func (s *taskStarter) StartGeneration(ctx context.Context, input chatWorkerTaskS chainModeDisabled = true } if err := s.waitGenerationRetry(ctx, decision.delay); err != nil { - return err + return xerrors.Errorf("wait generation retry: %w", err) } continue } @@ -475,13 +478,13 @@ func loadGenerationState( err := machine.ReadLock(ctx, func(store database.Store) error { chat, err := store.GetChatByID(ctx, input.ChatID) if errors.Is(err, sql.ErrNoRows) { - return errTaskExpectedExit + return errors.Join(errTaskExpectedExit, xerrors.Errorf("load locked chat: %w", err)) } if err != nil { return xerrors.Errorf("load locked chat: %w", err) } if err := verifyTaskFence(chat, input, database.ChatStatusRunning, taskFenceOptions{requireHistory: true}); err != nil { - return err + return xerrors.Errorf("verifyTaskFence: %w", err) } loaded, err := store.GetChatMessagesByChatID(ctx, database.GetChatMessagesByChatIDParams{ ChatID: input.ChatID, @@ -511,13 +514,13 @@ func (*taskStarter) recordGenerationRetry( err := machine.Update(ctx, func(tx *chatstate.Tx, store database.Store) error { locked, err := store.GetChatByID(ctx, input.ChatID) if errors.Is(err, sql.ErrNoRows) { - return errTaskExpectedExit + return errors.Join(errTaskExpectedExit, xerrors.Errorf("load chat: %w", err)) } if err != nil { return xerrors.Errorf("load chat: %w", err) } if err := verifyTaskFence(locked, input, database.ChatStatusRunning, taskFenceOptions{requireHistory: true}); err != nil { - return err + return xerrors.Errorf("verifyTaskFence: %w", err) } decision.generationAttempt = locked.GenerationAttempt if locked.GenerationAttempt <= 0 || locked.GenerationAttempt >= int64(chatretry.MaxAttempts) { @@ -544,7 +547,10 @@ func (*taskStarter) recordGenerationRetry( _, err = tx.RecordRetryState(chatstate.RecordRetryStateInput{ RetryState: pqtype.NullRawMessage{RawMessage: encoded, Valid: true}, }) - return err + if err != nil { + return xerrors.Errorf("record retry state: %w", err) + } + return nil }) if errors.Is(err, errRetryStateDecisionOnly) { return decision, nil @@ -643,7 +649,7 @@ func (s *taskStarter) generateAssistant( ) error { attempt, _, publish, closeEpisode, err := s.beginGenerationAttempt(ctx, machine, input) if err != nil { - return err + return xerrors.Errorf("begin generation attempt: %w", err) } defer closeEpisode() runCtx := input.DebugTurn.Ensure(ctx, prepared.Chat, prepared.Debug) @@ -663,7 +669,7 @@ func (s *taskStarter) generateAssistant( Metrics: s.server.metrics, }) if err != nil { - return err + return xerrors.Errorf("generate assistant: %w", err) } if len(outcome.Step.Content) == 0 { return s.finishGenerationTurn(ctx, machine, input, attempt, generationDecision{kind: generationActionFinishTurn, finishReason: generationFinishReasonComplete}, generationAttemptRequired) @@ -691,7 +697,7 @@ func (s *taskStarter) executeLocalTools( ) error { attempt, _, publish, closeEpisode, err := s.beginGenerationAttempt(ctx, machine, input) if err != nil { - return err + return xerrors.Errorf("beginGenerationAttempt: %w", err) } defer closeEpisode() provider := "" @@ -720,7 +726,7 @@ func (s *taskStarter) executeLocalTools( Clock: s.opts.Clock, }) if err != nil { - return err + return xerrors.Errorf("execute local tools: %w", err) } messages, err := buildCommitStepMessages(buildCommitStepMessagesInput{ modelConfigID: prepared.ModelConfigID, @@ -744,7 +750,7 @@ func (s *taskStarter) generateCompaction( ) error { attempt, _, publish, closeEpisode, err := s.beginGenerationAttempt(ctx, machine, input) if err != nil { - return err + return xerrors.Errorf("beginGenerationAttempt: %w", err) } defer closeEpisode() if prepared.Compaction == nil { @@ -755,7 +761,7 @@ func (s *taskStarter) generateCompaction( outcome, err := chatloop.GenerateCompaction(ctx, compactionOpts) if err != nil { s.server.metrics.RecordCompaction(compactionProvider(compactionOpts), compactionModel(compactionOpts), false, err) - return err + return xerrors.Errorf("generate compaction: %w", err) } if strings.TrimSpace(outcome.SystemSummary) == "" || strings.TrimSpace(outcome.SummaryReport) == "" { err := xerrors.New("compaction produced no summary") @@ -779,7 +785,10 @@ func (s *taskStarter) generateCompaction( VisibleIndexes: visibleMessageIndexes(messages.Messages), }) s.server.metrics.RecordCompaction(compactionProvider(compactionOpts), compactionModel(compactionOpts), err == nil, err) - return err + if err != nil { + return xerrors.Errorf("commit generation step: %w", err) + } + return nil } func compactionProvider(opts chatloop.GenerateCompactionOptions) string { @@ -809,9 +818,6 @@ func (s *taskStarter) persistWorkspaceContext( input chatWorkerTaskStartInput, locked database.Chat, ) error { - if s.server == nil { - return errTaskExpectedExit - } messages, err := s.opts.Store.GetChatMessagesByChatID(ctx, database.GetChatMessagesByChatIDParams{ ChatID: input.ChatID, AfterID: 0, @@ -821,7 +827,7 @@ func (s *taskStarter) persistWorkspaceContext( } attempt, _, _, closeEpisode, err := s.beginGenerationAttempt(ctx, machine, input) if err != nil { - return err + return xerrors.Errorf("beginGenerationAttempt: %w", err) } defer closeEpisode() modelOpts := modelBuildOptionsFromMessages(messages) @@ -880,17 +886,17 @@ func (s *taskStarter) beginGenerationAttempt( err := machine.Update(ctx, func(tx *chatstate.Tx, store database.Store) error { locked, err := store.GetChatByID(ctx, input.ChatID) if errors.Is(err, sql.ErrNoRows) { - return errTaskExpectedExit + return errors.Join(errTaskExpectedExit, xerrors.Errorf("load chat: %w", err)) } if err != nil { return xerrors.Errorf("load chat: %w", err) } if err := verifyTaskFence(locked, input, database.ChatStatusRunning, taskFenceOptions{requireHistory: true}); err != nil { - return err + return xerrors.Errorf("verifyTaskFence: %w", err) } result, err := tx.RecordGenerationAttempt(chatstate.RecordGenerationAttemptInput{}) if err != nil { - return err + return xerrors.Errorf("tx.RecordGenerationAttempt: %w", err) } attempt = result.GenerationAttempt committed, err = store.GetChatByID(ctx, input.ChatID) @@ -935,17 +941,17 @@ func (s *taskStarter) commitGenerationStep( err := machine.Update(ctx, func(tx *chatstate.Tx, store database.Store) error { locked, err := store.GetChatByID(ctx, input.ChatID) if errors.Is(err, sql.ErrNoRows) { - return errTaskExpectedExit + return errors.Join(errTaskExpectedExit, xerrors.Errorf("load chat: %w", err)) } if err != nil { return xerrors.Errorf("load chat: %w", err) } if err := verifyGenerationFence(locked, input, attempt); err != nil { - return err + return xerrors.Errorf("verifyGenerationFence: %w", err) } commitResult, err := tx.CommitStep(chatstate.CommitStepInput{Messages: messages.Messages}) if err != nil { - return err + return xerrors.Errorf("tx.CommitStep: %w", err) } insertedMessages = make([]runnerActionMessage, 0, len(commitResult.InsertedMessages)) for _, msg := range commitResult.InsertedMessages { @@ -977,16 +983,16 @@ func (s *taskStarter) enterRequiresAction( err := machine.Update(ctx, func(tx *chatstate.Tx, store database.Store) error { locked, err := store.GetChatByID(ctx, input.ChatID) if errors.Is(err, sql.ErrNoRows) { - return errTaskExpectedExit + return errors.Join(errTaskExpectedExit, xerrors.Errorf("load chat: %w", err)) } if err != nil { return xerrors.Errorf("load chat: %w", err) } if err := verifyTaskFence(locked, input, database.ChatStatusRunning, taskFenceOptions{requireHistory: true}); err != nil { - return err + return xerrors.Errorf("verifyTaskFence: %w", err) } if _, err := tx.EnterRequiresAction(chatstate.EnterRequiresActionInput{}); err != nil { - return err + return xerrors.Errorf("tx.EnterRequiresAction: %w", err) } committed, err = store.GetChatByID(ctx, input.ChatID) if err != nil { @@ -998,7 +1004,7 @@ func (s *taskStarter) enterRequiresAction( return normalizeTaskTransitionError(err, "enter requires action") } if err := s.publishWatchAndRoute(ctx, committed, codersdk.ChatWatchEventKindActionRequired); err != nil { - return err + return xerrors.Errorf("publish watch and route: %w", err) } return s.afterGenerationOutcome(ctx, generationOutcome{ Chat: committed, @@ -1026,21 +1032,21 @@ func (s *taskStarter) finishGenerationTurn( err := machine.Update(ctx, func(tx *chatstate.Tx, store database.Store) error { locked, err := store.GetChatByID(ctx, input.ChatID) if errors.Is(err, sql.ErrNoRows) { - return errTaskExpectedExit + return errors.Join(errTaskExpectedExit, xerrors.Errorf("load chat: %w", err)) } if err != nil { return xerrors.Errorf("load chat: %w", err) } if attemptFence == generationAttemptRequired { if err := verifyGenerationFence(locked, input, attempt); err != nil { - return err + return xerrors.Errorf("verifyGenerationFence: %w", err) } } else if err := verifyTaskFence(locked, input, database.ChatStatusRunning, taskFenceOptions{requireHistory: true}); err != nil { - return err + return xerrors.Errorf("verifyTaskFence: %w", err) } finishResult, err := tx.FinishTurn(chatstate.FinishTurnInput{}) if err != nil { - return err + return xerrors.Errorf("tx.FinishTurn: %w", err) } if finishResult.PromotedMessage != nil { decision.promotedMessageID = finishResult.PromotedMessage.ID @@ -1055,7 +1061,7 @@ func (s *taskStarter) finishGenerationTurn( watchCtx, cancel := context.WithTimeout(context.WithoutCancel(ctx), postCommitWatchPublishTimeout) defer cancel() if err := s.publishWatchWithRetry(watchCtx, committed, codersdk.ChatWatchEventKindStatusChange); err != nil { - return err + return xerrors.Errorf("publish watch and route: %w", err) } if err := s.afterGenerationOutcome(ctx, generationOutcome{ Chat: committed, @@ -1063,7 +1069,7 @@ func (s *taskStarter) finishGenerationTurn( WatchEventKind: codersdk.ChatWatchEventKindStatusChange, PromotedMessageID: decision.promotedMessageID, }); err != nil { - return err + return xerrors.Errorf("after generation outcome: %w", err) } s.routeStateHint(ctx, stateUpdateFromChat(committed)) return nil @@ -1096,20 +1102,20 @@ func (s *taskStarter) finishGenerationError( err := machine.Update(ctx, func(tx *chatstate.Tx, store database.Store) error { locked, err := store.GetChatByID(ctx, input.ChatID) if errors.Is(err, sql.ErrNoRows) { - return errTaskExpectedExit + return errors.Join(errTaskExpectedExit, xerrors.Errorf("load chat: %w", err)) } if err != nil { return xerrors.Errorf("load chat: %w", err) } if attemptFence == generationAttemptRequired { if err := verifyGenerationFence(locked, input, attempt); err != nil { - return err + return xerrors.Errorf("verifyGenerationFence: %w", err) } } else if err := verifyTaskFence(locked, input, database.ChatStatusRunning, taskFenceOptions{requireHistory: true}); err != nil { - return err + return xerrors.Errorf("verifyTaskFence: %w", err) } if _, err := tx.FinishError(chatstate.FinishErrorInput{LastError: lastError}); err != nil { - return err + return xerrors.Errorf("tx.FinishError: %w", err) } committed, err = store.GetChatByID(ctx, input.ChatID) if err != nil { @@ -1122,7 +1128,7 @@ func (s *taskStarter) finishGenerationError( } input.DebugTurn.RecordOutcome(chatdebug.StatusError) if err := s.publishWatchAndRoute(ctx, committed, codersdk.ChatWatchEventKindStatusChange); err != nil { - return err + return xerrors.Errorf("publish watch and route: %w", err) } return s.afterGenerationOutcome(ctx, generationOutcome{ Chat: committed, @@ -1160,10 +1166,10 @@ func (s *taskStarter) afterGenerationOutcome(ctx context.Context, outcome genera func verifyGenerationFence(chat database.Chat, input chatWorkerTaskStartInput, attempt int64) error { if err := verifyTaskFence(chat, input, database.ChatStatusRunning, taskFenceOptions{requireHistory: true}); err != nil { - return err + return xerrors.Errorf("verifyTaskFence: %w", err) } if chat.GenerationAttempt != attempt { - return errTaskExpectedExit + return errors.Join(errTaskExpectedExit, xerrors.Errorf("generation fence mismatch: %d != %d", chat.GenerationAttempt, attempt)) } return nil } diff --git a/coderd/x/chatd/runner.go b/coderd/x/chatd/runner.go index 38edc7249d..651705df16 100644 --- a/coderd/x/chatd/runner.go +++ b/coderd/x/chatd/runner.go @@ -245,7 +245,12 @@ func (r *runner) runTask( done chan<- struct{}, ) { defer close(done) - err := runTaskWithRetry(ctx, r.opts.retryOptions(), kind, func(ctx context.Context) error { + taskInfo := retryWrapperTaskInfo{ + ChatID: input.ChatID, + WorkerID: input.WorkerID, + RunnerID: input.RunnerID, + } + err := runTaskWithRetry(ctx, r.opts.retryOptions(), kind, taskInfo, func(ctx context.Context) error { unlock, ok := r.localLocks.acquire(ctx, key) if !ok { return errors.Join(errTaskExpectedExit, xerrors.Errorf("runTask acquire local lock: %w", ctx.Err())) diff --git a/coderd/x/chatd/tasks.go b/coderd/x/chatd/tasks.go index f370ea852f..82a4ada1e3 100644 --- a/coderd/x/chatd/tasks.go +++ b/coderd/x/chatd/tasks.go @@ -61,6 +61,12 @@ type retryWrapperOptions struct { maxDelay time.Duration } +type retryWrapperTaskInfo struct { + ChatID uuid.UUID + WorkerID uuid.UUID + RunnerID uuid.UUID +} + // 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: @@ -69,6 +75,7 @@ func runTaskWithRetry( ctx context.Context, opts retryWrapperOptions, kind taskKind, + info retryWrapperTaskInfo, fn func(context.Context) error, ) error { if opts.clock == nil { @@ -113,6 +120,9 @@ func runTaskWithRetry( opts.logger.Debug(ctx, "chatworker task exited", slog.F("task_kind", kind), slog.F("reason", exitReason), + slog.F("chat_id", info.ChatID), + slog.F("worker_id", info.WorkerID), + slog.F("runner_id", info.RunnerID), slogError(err), ) return nil @@ -121,6 +131,9 @@ func runTaskWithRetry( opts.logger.Warn(ctx, "chatworker task retrying", slog.F("task_kind", kind), slog.F("delay", delay), + slog.F("chat_id", info.ChatID), + slog.F("worker_id", info.WorkerID), + slog.F("runner_id", info.RunnerID), slogError(err), ) timer := opts.clock.NewTimer(delay, "chatworker", "task-retry-"+string(kind)) @@ -230,13 +243,13 @@ func (s *taskStarter) StartInterrupt(ctx context.Context, input chatWorkerTaskSt err := machine.ReadLock(ctx, func(store database.Store) error { locked, err := store.GetChatByID(ctx, input.ChatID) if errors.Is(err, sql.ErrNoRows) { - return errTaskExpectedExit + return errors.Join(errTaskExpectedExit, xerrors.Errorf("load locked chat: %w", err)) } if err != nil { return xerrors.Errorf("load locked chat: %w", err) } if err := verifyTaskFence(locked, input, database.ChatStatusInterrupting, taskFenceOptions{requireHistory: true}); err != nil { - return err + return xerrors.Errorf("verifyTaskFence: %w", err) } chat = locked return nil @@ -282,24 +295,24 @@ func (s *taskStarter) StartInterrupt(ctx context.Context, input chatWorkerTaskSt err = machine.Update(ctx, func(tx *chatstate.Tx, store database.Store) error { locked, err := store.GetChatByID(ctx, input.ChatID) if errors.Is(err, sql.ErrNoRows) { - return errTaskExpectedExit + return errors.Join(errTaskExpectedExit, xerrors.Errorf("load chat: %w", err)) } if err != nil { return xerrors.Errorf("load chat: %w", err) } if err := verifyTaskFence(locked, input, database.ChatStatusInterrupting, taskFenceOptions{requireHistory: true}); err != nil { - return err + return xerrors.Errorf("verifyTaskFence: %w", err) } messages := partialMessages committedCancels, err := committedPendingLocalToolCancellationMessages(ctx, store, locked, s.opts.Clock.Now("chatworker", "interrupt")) if err != nil { - return err + return xerrors.Errorf("committed pending local tool cancellation messages: %w", err) } if len(committedCancels) > 0 { messages = append(append([]chatstate.Message{}, partialMessages...), committedCancels...) } if _, err := tx.FinishInterruption(chatstate.FinishInterruptionInput{PartialMessages: messages}); err != nil { - return err + return xerrors.Errorf("finish interruption: %w", err) } committed, err = store.GetChatByID(ctx, input.ChatID) if err != nil { @@ -315,7 +328,7 @@ func (s *taskStarter) StartInterrupt(ctx context.Context, input chatWorkerTaskSt } input.DebugTurn.RecordOutcome(chatdebug.StatusInterrupted) if err := s.publishWatchAndRoute(ctx, committed, codersdk.ChatWatchEventKindStatusChange); err != nil { - return err + return xerrors.Errorf("publish watch and route: %w", err) } return s.runAfterInterruptionOutcome(ctx, interruptionOutcome{ Chat: committed, @@ -343,16 +356,16 @@ func (s *taskStarter) StartRequiresActionTimeout(ctx context.Context, input chat for { decision, err := decideRequiresActionTimeout(ctx, machine, input) if err != nil { - return err + return xerrors.Errorf("decide requires action timeout: %w", err) } if decision.cancel { return s.cancelRequiresAction(ctx, machine, input, decision.reason) } if !decision.waitUntil.Valid { - return errTaskExpectedExit + return errors.Join(errTaskExpectedExit, xerrors.Errorf("requires action deadline is missing")) } if err := s.waitUntil(ctx, decision.waitUntil.Time); err != nil { - return err + return xerrors.Errorf("wait until: %w", err) } } } @@ -372,13 +385,13 @@ func decideRequiresActionTimeout( err := machine.ReadLock(ctx, func(store database.Store) error { locked, err := store.GetChatByID(ctx, input.ChatID) if errors.Is(err, sql.ErrNoRows) { - return errTaskExpectedExit + return errors.Join(errTaskExpectedExit, xerrors.Errorf("load locked chat: %w", err)) } if err != nil { return xerrors.Errorf("load locked chat: %w", err) } if err := verifyTaskFence(locked, input, database.ChatStatusRequiresAction, taskFenceOptions{requireHistory: true}); err != nil { - return err + return xerrors.Errorf("verifyTaskFence: %w", err) } if !locked.RequiresActionDeadlineAt.Valid { decision.cancel = true @@ -428,13 +441,13 @@ func (s *taskStarter) cancelRequiresAction( err := machine.Update(ctx, func(tx *chatstate.Tx, store database.Store) error { locked, err := store.GetChatByID(ctx, input.ChatID) if errors.Is(err, sql.ErrNoRows) { - return errTaskExpectedExit + return errors.Join(errTaskExpectedExit, xerrors.Errorf("load locked chat: %w", err)) } if err != nil { return xerrors.Errorf("load chat: %w", err) } if err := verifyTaskFence(locked, input, database.ChatStatusRequiresAction, taskFenceOptions{requireHistory: true}); err != nil { - return err + return xerrors.Errorf("verifyTaskFence: %w", err) } if locked.RequiresActionDeadlineAt.Valid { now, err := store.GetDatabaseNow(ctx) @@ -442,11 +455,11 @@ func (s *taskStarter) cancelRequiresAction( return xerrors.Errorf("get database time: %w", err) } if now.Before(locked.RequiresActionDeadlineAt.Time) { - return errTaskExpectedExit + return errors.Join(errTaskExpectedExit, xerrors.Errorf("requires action deadline is in the future")) } } if _, err := tx.CancelRequiresAction(chatstate.CancelRequiresActionInput{Reason: reason}); err != nil { - return err + return xerrors.Errorf("cancel requires action: %w", err) } committed, err = store.GetChatByID(ctx, input.ChatID) if err != nil { @@ -470,20 +483,20 @@ func (s *taskStarter) StartAbandon(ctx context.Context, input chatWorkerTaskStar locked, err := store.GetChatByID(ctx, input.ChatID) if errors.Is(err, sql.ErrNoRows) { mismatch = true - return errTaskExpectedExit + return errors.Join(errTaskExpectedExit, xerrors.Errorf("load chat: %w", err)) } if err != nil { return xerrors.Errorf("load chat: %w", err) } if !ownedByTask(locked, input) { mismatch = true - return errTaskExpectedExit + return errors.Join(errTaskExpectedExit, xerrors.Errorf("chat not owned by task")) } if err := verifyTaskFence(locked, input, input.Status, taskFenceOptions{requireHistory: true, allowArchived: true}); err != nil { - return err + return xerrors.Errorf("task fence mismatch: %w", err) } if _, err := tx.Abandon(chatstate.AbandonInput{}); err != nil { - return err + return xerrors.Errorf("abandon chat: %w", err) } return nil }) @@ -527,7 +540,7 @@ func (s *taskStarter) publishWatchAndRoute( watchCtx, cancel := context.WithTimeout(context.WithoutCancel(ctx), postCommitWatchPublishTimeout) defer cancel() if err := s.publishWatchWithRetry(watchCtx, chat, kind); err != nil { - return err + return xerrors.Errorf("publish watch with retry: %w", err) } s.routeStateHint(ctx, stateUpdateFromChat(chat)) return nil @@ -589,16 +602,16 @@ func verifyTaskFence( opts taskFenceOptions, ) error { if !ownedByTask(chat, input) { - return errTaskExpectedExit + return errors.Join(errTaskExpectedExit, xerrors.Errorf("chat not owned by task")) } if chat.Status != status { - return errTaskExpectedExit + return errors.Join(errTaskExpectedExit, xerrors.Errorf("chat status mismatch: %s != %s", chat.Status, status)) } if !opts.allowArchived && chat.Archived { - return errTaskExpectedExit + return errors.Join(errTaskExpectedExit, xerrors.Errorf("chat archived")) } if opts.requireHistory && chat.HistoryVersion != input.HistoryVersion { - return errTaskExpectedExit + return errors.Join(errTaskExpectedExit, xerrors.Errorf("chat history version mismatch: %d != %d", chat.HistoryVersion, input.HistoryVersion)) } return nil } diff --git a/coderd/x/chatd/tasks_test.go b/coderd/x/chatd/tasks_test.go index 8fcc588c35..b2115dfe9a 100644 --- a/coderd/x/chatd/tasks_test.go +++ b/coderd/x/chatd/tasks_test.go @@ -42,7 +42,7 @@ func TestRetryWrapper_ExpectedExitsDoNotRetry(t *testing.T) { logger: sink.Logger(), initialDelay: time.Second, maxDelay: time.Second, - }, taskKindInterrupt, func(context.Context) error { + }, taskKindInterrupt, retryWrapperTaskInfo{}, func(context.Context) error { calls++ return errTaskExpectedExit }) @@ -67,7 +67,7 @@ func TestRetryWrapper_UnexpectedErrorsRetry(t *testing.T) { logger: sink.Logger(), initialDelay: time.Minute, maxDelay: time.Minute, - }, taskKindRequiresActionTimeout, func(context.Context) error { + }, taskKindRequiresActionTimeout, retryWrapperTaskInfo{}, func(context.Context) error { calls++ if calls == 1 { return xerrors.New("database unavailable") @@ -103,7 +103,7 @@ func TestRetryWrapper_PanicsRetry(t *testing.T) { logger: sink.Logger(), initialDelay: time.Minute, maxDelay: time.Minute, - }, taskKindGeneration, func(context.Context) error { + }, taskKindGeneration, retryWrapperTaskInfo{}, func(context.Context) error { calls++ if calls == 1 { panic("database unavailable") @@ -146,7 +146,7 @@ func TestRetryWrapper_TaskTimeoutDBQueryCancellationRetries(t *testing.T) { logger: sink.Logger(), initialDelay: time.Minute, maxDelay: time.Minute, - }, taskKindGeneration, func(ctx context.Context) error { + }, taskKindGeneration, retryWrapperTaskInfo{}, func(ctx context.Context) error { calls++ if calls == 1 { close(firstCallStarted) @@ -190,7 +190,7 @@ func TestRetryWrapper_ContextCancellationDoesNotRetryOrLog(t *testing.T) { logger: sink.Logger(), initialDelay: time.Second, maxDelay: time.Second, - }, taskKindGeneration, func(context.Context) error { + }, taskKindGeneration, retryWrapperTaskInfo{}, func(context.Context) error { calls++ return original })