opensource/coder
3
0
Fork 0
mirror of https://github.com/coder/coder.git synced 2026-09-24 15:04:27 +08:00
Code Issues Packages Projects Releases Wiki Activity

chore(coderd/x/chatd): improve runner logging (#26522)

Browse Source 
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.
This commit is contained in:
Hugo Dutka
2026-06-18 19:02:42 +02:00
committed by GitHub
parent 023a1c992c
commit c0c0d1e353
4 changed files with 99 additions and 75 deletions
Download Patch File Download Diff File Expand all files Collapse all files
coderd/x/chatd/generation.go
+50 -44
View File
@@ -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
}
coderd/x/chatd/runner.go
+6 -1
View File
@@ -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()))
coderd/x/chatd/tasks.go
+38 -25
View File
@@ -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
}
coderd/x/chatd/tasks_test.go
+5 -5
View File
@@ -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
})