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

fix(coderd/x/chatd/chatloop): use stream silence timeout (#25782)

Browse Source 
Replaces the 60 second first-token timeout in the chat loop with a 10
minute stream-silence timeout.

Previously, the guard bounded only the gap before the first stream part.
Once any part arrived the attempt could hang indefinitely if the
provider stopped streaming without closing the connection, and even
normal long-running responses could be killed after 60 seconds if the
provider was slow to emit the first token.

The guard now arms when a model attempt opens its stream, resets on
every received stream part, and fires after 10 minutes of complete
silence. The existing retry path still handles the timeout, and the
public `startup_timeout` error kind is preserved to avoid API and
frontend churn.

10 minutes matches the default request timeout used by the Anthropic and
OpenAI Python SDKs.


Closes CODAGT-493
This commit is contained in:
Ethan
2026-05-28 21:02:40 +10:00
committed by GitHub
parent 3770176b7f
commit 7e2f7198dd
2 changed files with 265 additions and 101 deletions
Download Patch File Download Diff File Expand all files Collapse all files
coderd/x/chatd/chatloop/chatloop_run_internal_test.go
+190 -57
View File
@@ -581,13 +581,13 @@ func TestRun_OnRetryEnrichesProvider(t *testing.T) {
)
}
func TestStartupGuard_DisarmAndFireRace(t *testing.T) {
func TestStreamSilenceGuard_DisarmAndFireRace(t *testing.T) {
t.Parallel()
for range 128 {
var cancels atomic.Int32
guard := newStartupGuard(quartz.NewReal(), time.Hour, func(err error) {
if errors.Is(err, errStartupTimeout) {
guard := newStreamSilenceGuard(quartz.NewReal(), time.Hour, func(err error) {
if errors.Is(err, errStreamSilenceTimeout) {
cancels.Add(1)
}
})
@@ -618,17 +618,17 @@ func TestStartupGuard_DisarmAndFireRace(t *testing.T) {
}
}
func TestStartupGuard_DisarmPreservesPermanentError(t *testing.T) {
func TestStreamSilenceGuard_DisarmPreservesPermanentError(t *testing.T) {
t.Parallel()
attemptCtx, cancelAttempt := context.WithCancelCause(context.Background())
defer cancelAttempt(nil)
guard := newStartupGuard(quartz.NewReal(), time.Hour, cancelAttempt)
guard := newStreamSilenceGuard(quartz.NewReal(), time.Hour, cancelAttempt)
guard.Disarm()
guard.onTimeout()
classified := chaterror.Classify(classifyStartupTimeout(
classified := chaterror.Classify(classifyStreamSilenceTimeout(
attemptCtx,
"openai",
xerrors.New("invalid model"),
@@ -638,10 +638,10 @@ func TestStartupGuard_DisarmPreservesPermanentError(t *testing.T) {
require.Nil(t, context.Cause(attemptCtx))
}
func TestRun_RetriesStartupTimeoutWhileOpeningStream(t *testing.T) {
func TestRun_RetriesSilenceTimeoutWhileOpeningStream(t *testing.T) {
t.Parallel()
const startupTimeout = 5 * time.Millisecond
const silenceTimeout = 5 * time.Millisecond
ctx, cancel := context.WithTimeout(
context.Background(),
@@ -650,7 +650,7 @@ func TestRun_RetriesStartupTimeoutWhileOpeningStream(t *testing.T) {
defer cancel()
mClock := quartz.NewMock(t)
trap := mClock.Trap().AfterFunc("startupGuard")
trap := mClock.Trap().AfterFunc(streamSilenceGuardTimerTag)
defer trap.Close()
attempts := 0
@@ -675,10 +675,10 @@ func TestRun_RetriesStartupTimeoutWhileOpeningStream(t *testing.T) {
done := make(chan error, 1)
go func() {
done <- Run(context.Background(), RunOptions{
Model: model,
MaxSteps: 1,
StartupTimeout: startupTimeout,
Clock: mClock,
Model: model,
MaxSteps: 1,
StreamSilenceTimeout: silenceTimeout,
Clock: mClock,
PersistStep: func(_ context.Context, _ PersistedStep) error {
return nil
},
@@ -694,7 +694,7 @@ func TestRun_RetriesStartupTimeoutWhileOpeningStream(t *testing.T) {
}()
trap.MustWait(ctx).MustRelease(ctx)
mClock.Advance(startupTimeout).MustWait(ctx)
mClock.Advance(silenceTimeout).MustWait(ctx)
trap.MustWait(ctx).MustRelease(ctx)
require.NoError(t, awaitRunResult(ctx, t, done))
@@ -710,9 +710,9 @@ func TestRun_RetriesStartupTimeoutWhileOpeningStream(t *testing.T) {
)
select {
case cause := <-attemptCause:
require.ErrorIs(t, cause, errStartupTimeout)
require.ErrorIs(t, cause, errStreamSilenceTimeout)
case <-ctx.Done():
t.Fatal("timed out waiting for startup timeout cause")
t.Fatal("timed out waiting for silence timeout cause")
}
}
@@ -728,7 +728,7 @@ func TestRun_HTTP2TransportErrorClassifiedAsRetryableTimeout(t *testing.T) {
t.Run(provider, func(t *testing.T) {
t.Parallel()
const startupTimeout = 5 * time.Millisecond
const silenceTimeout = 5 * time.Millisecond
ctx, cancel := context.WithTimeout(
context.Background(),
@@ -737,7 +737,7 @@ func TestRun_HTTP2TransportErrorClassifiedAsRetryableTimeout(t *testing.T) {
defer cancel()
mClock := quartz.NewMock(t)
trap := mClock.Trap().AfterFunc("startupGuard")
trap := mClock.Trap().AfterFunc(streamSilenceGuardTimerTag)
defer trap.Close()
attempts := 0
@@ -763,10 +763,10 @@ func TestRun_HTTP2TransportErrorClassifiedAsRetryableTimeout(t *testing.T) {
done := make(chan error, 1)
go func() {
done <- Run(context.Background(), RunOptions{
Model: model,
MaxSteps: 1,
StartupTimeout: startupTimeout,
Clock: mClock,
Model: model,
MaxSteps: 1,
StreamSilenceTimeout: silenceTimeout,
Clock: mClock,
PersistStep: func(_ context.Context, _ PersistedStep) error {
return nil
},
@@ -795,10 +795,10 @@ func TestRun_HTTP2TransportErrorClassifiedAsRetryableTimeout(t *testing.T) {
}
}
func TestRun_RetriesStartupTimeoutBeforeFirstPart(t *testing.T) {
func TestRun_RetriesSilenceTimeoutBeforeFirstPart(t *testing.T) {
t.Parallel()
const startupTimeout = 5 * time.Millisecond
const silenceTimeout = 5 * time.Millisecond
ctx, cancel := context.WithTimeout(
context.Background(),
@@ -807,7 +807,7 @@ func TestRun_RetriesStartupTimeoutBeforeFirstPart(t *testing.T) {
defer cancel()
mClock := quartz.NewMock(t)
trap := mClock.Trap().AfterFunc("startupGuard")
trap := mClock.Trap().AfterFunc(streamSilenceGuardTimerTag)
defer trap.Close()
attempts := 0
@@ -837,10 +837,10 @@ func TestRun_RetriesStartupTimeoutBeforeFirstPart(t *testing.T) {
done := make(chan error, 1)
go func() {
done <- Run(context.Background(), RunOptions{
Model: model,
MaxSteps: 1,
StartupTimeout: startupTimeout,
Clock: mClock,
Model: model,
MaxSteps: 1,
StreamSilenceTimeout: silenceTimeout,
Clock: mClock,
PersistStep: func(_ context.Context, _ PersistedStep) error {
return nil
},
@@ -856,7 +856,7 @@ func TestRun_RetriesStartupTimeoutBeforeFirstPart(t *testing.T) {
}()
trap.MustWait(ctx).MustRelease(ctx)
mClock.Advance(startupTimeout).MustWait(ctx)
mClock.Advance(silenceTimeout).MustWait(ctx)
trap.MustWait(ctx).MustRelease(ctx)
require.NoError(t, awaitRunResult(ctx, t, done))
@@ -872,16 +872,16 @@ func TestRun_RetriesStartupTimeoutBeforeFirstPart(t *testing.T) {
)
select {
case cause := <-attemptCause:
require.ErrorIs(t, cause, errStartupTimeout)
require.ErrorIs(t, cause, errStreamSilenceTimeout)
case <-ctx.Done():
t.Fatal("timed out waiting for startup timeout cause")
t.Fatal("timed out waiting for silence timeout cause")
}
}
func TestRun_FirstPartDisarmsStartupTimeout(t *testing.T) {
func TestRun_StreamPartsResetSilenceTimeout(t *testing.T) {
t.Parallel()
const startupTimeout = 5 * time.Millisecond
const silenceTimeout = 5 * time.Millisecond
ctx, cancel := context.WithTimeout(
context.Background(),
@@ -890,12 +890,17 @@ func TestRun_FirstPartDisarmsStartupTimeout(t *testing.T) {
defer cancel()
mClock := quartz.NewMock(t)
trap := mClock.Trap().AfterFunc("startupGuard")
armTrap := mClock.Trap().AfterFunc(streamSilenceGuardTimerTag)
defer armTrap.Close()
resetTrap := mClock.Trap().TimerReset(streamSilenceGuardTimerTag)
defer resetTrap.Close()
attempts := 0
retried := false
firstPartYielded := make(chan struct{}, 1)
continueStream := make(chan struct{})
secondPartYielded := make(chan struct{}, 1)
continueToSecond := make(chan struct{})
continueToFinish := make(chan struct{})
model := &chattest.FakeModel{
ProviderName: "openai",
StreamFn: func(ctx context.Context, _ fantasy.Call) (fantasy.StreamResponse, error) {
@@ -910,7 +915,29 @@ func TestRun_FirstPartDisarmsStartupTimeout(t *testing.T) {
}
select {
case <-continueStream:
case <-continueToSecond:
case <-ctx.Done():
_ = yield(fantasy.StreamPart{
Type: fantasy.StreamPartTypeError,
Error: ctx.Err(),
})
return
}
if !yield(fantasy.StreamPart{
Type: fantasy.StreamPartTypeTextDelta,
ID: "text-1",
Delta: "done",
}) {
return
}
select {
case secondPartYielded <- struct{}{}:
default:
}
select {
case <-continueToFinish:
case <-ctx.Done():
_ = yield(fantasy.StreamPart{
Type: fantasy.StreamPartTypeError,
@@ -920,7 +947,6 @@ func TestRun_FirstPartDisarmsStartupTimeout(t *testing.T) {
}
parts := []fantasy.StreamPart{
{Type: fantasy.StreamPartTypeTextDelta, ID: "text-1", Delta: "done"},
{Type: fantasy.StreamPartTypeTextEnd, ID: "text-1"},
{Type: fantasy.StreamPartTypeFinish, FinishReason: fantasy.FinishReasonStop},
}
@@ -936,10 +962,10 @@ func TestRun_FirstPartDisarmsStartupTimeout(t *testing.T) {
done := make(chan error, 1)
go func() {
done <- Run(context.Background(), RunOptions{
Model: model,
MaxSteps: 1,
StartupTimeout: startupTimeout,
Clock: mClock,
Model: model,
MaxSteps: 1,
StreamSilenceTimeout: silenceTimeout,
Clock: mClock,
PersistStep: func(_ context.Context, _ PersistedStep) error {
return nil
},
@@ -954,23 +980,130 @@ func TestRun_FirstPartDisarmsStartupTimeout(t *testing.T) {
})
}()
trap.MustWait(ctx).MustRelease(ctx)
trap.Close()
armTrap.MustWait(ctx).MustRelease(ctx)
resetTrap.MustWait(ctx).MustRelease(ctx)
select {
case <-firstPartYielded:
case <-ctx.Done():
t.Fatal("timed out waiting for first stream part")
}
mClock.Advance(startupTimeout).MustWait(ctx)
close(continueStream)
mClock.Advance(silenceTimeout / 2).MustWait(ctx)
close(continueToSecond)
resetTrap.MustWait(ctx).MustRelease(ctx)
select {
case <-secondPartYielded:
case <-ctx.Done():
t.Fatal("timed out waiting for second stream part")
}
mClock.Advance(silenceTimeout / 2).MustWait(ctx)
close(continueToFinish)
resetTrap.MustWait(ctx).MustRelease(ctx)
resetTrap.MustWait(ctx).MustRelease(ctx)
require.NoError(t, awaitRunResult(ctx, t, done))
require.Equal(t, 1, attempts)
require.False(t, retried)
}
func TestRun_RetriesSilenceTimeoutBetweenParts(t *testing.T) {
t.Parallel()
const silenceTimeout = 5 * time.Millisecond
ctx, cancel := context.WithTimeout(
context.Background(),
testutil.WaitLong,
)
defer cancel()
mClock := quartz.NewMock(t)
armTrap := mClock.Trap().AfterFunc(streamSilenceGuardTimerTag)
defer armTrap.Close()
resetTrap := mClock.Trap().TimerReset(streamSilenceGuardTimerTag)
defer resetTrap.Close()
attempts := 0
firstPartYielded := make(chan struct{}, 1)
attemptCause := make(chan error, 1)
var retries []chatretry.ClassifiedError
model := &chattest.FakeModel{
ProviderName: "openai",
StreamFn: func(ctx context.Context, _ fantasy.Call) (fantasy.StreamResponse, error) {
attempts++
if attempts == 1 {
return iter.Seq[fantasy.StreamPart](func(yield func(fantasy.StreamPart) bool) {
if !yield(fantasy.StreamPart{Type: fantasy.StreamPartTypeTextStart, ID: "text-1"}) {
return
}
select {
case firstPartYielded <- struct{}{}:
default:
}
<-ctx.Done()
attemptCause <- context.Cause(ctx)
_ = yield(fantasy.StreamPart{
Type: fantasy.StreamPartTypeError,
Error: ctx.Err(),
})
}), nil
}
return streamFromParts([]fantasy.StreamPart{{
Type: fantasy.StreamPartTypeFinish,
FinishReason: fantasy.FinishReasonStop,
}}), nil
},
}
done := make(chan error, 1)
go func() {
done <- Run(context.Background(), RunOptions{
Model: model,
MaxSteps: 1,
StreamSilenceTimeout: silenceTimeout,
Clock: mClock,
PersistStep: func(_ context.Context, _ PersistedStep) error {
return nil
},
OnRetry: func(
_ int,
_ error,
classified chatretry.ClassifiedError,
_ time.Duration,
) {
retries = append(retries, classified)
},
})
}()
armTrap.MustWait(ctx).MustRelease(ctx)
resetTrap.MustWait(ctx).MustRelease(ctx)
select {
case <-firstPartYielded:
case <-ctx.Done():
t.Fatal("timed out waiting for first stream part")
}
mClock.Advance(silenceTimeout).MustWait(ctx)
armTrap.MustWait(ctx).MustRelease(ctx)
resetTrap.MustWait(ctx).MustRelease(ctx)
require.NoError(t, awaitRunResult(ctx, t, done))
require.Equal(t, 2, attempts)
require.Len(t, retries, 1)
require.Equal(t, codersdk.ChatErrorKindStartupTimeout, retries[0].Kind)
require.True(t, retries[0].Retryable)
require.Equal(t, "openai", retries[0].Provider)
select {
case cause := <-attemptCause:
require.ErrorIs(t, cause, errStreamSilenceTimeout)
case <-ctx.Done():
t.Fatal("timed out waiting for silence timeout cause")
}
}
func TestRun_PanicInPublishMessagePartReleasesAttempt(t *testing.T) {
t.Parallel()
@@ -1014,10 +1147,10 @@ func TestRun_PanicInPublishMessagePartReleasesAttempt(t *testing.T) {
t.Fatal("expected Run to panic")
}
func TestRun_RetriesStartupTimeoutWhenStreamClosesSilently(t *testing.T) {
func TestRun_RetriesSilenceTimeoutWhenStreamStaysSilent(t *testing.T) {
t.Parallel()
const startupTimeout = 5 * time.Millisecond
const silenceTimeout = 5 * time.Millisecond
ctx, cancel := context.WithTimeout(
context.Background(),
@@ -1026,7 +1159,7 @@ func TestRun_RetriesStartupTimeoutWhenStreamClosesSilently(t *testing.T) {
defer cancel()
mClock := quartz.NewMock(t)
trap := mClock.Trap().AfterFunc("startupGuard")
trap := mClock.Trap().AfterFunc(streamSilenceGuardTimerTag)
defer trap.Close()
attempts := 0
@@ -1052,10 +1185,10 @@ func TestRun_RetriesStartupTimeoutWhenStreamClosesSilently(t *testing.T) {
done := make(chan error, 1)
go func() {
done <- Run(context.Background(), RunOptions{
Model: model,
MaxSteps: 1,
StartupTimeout: startupTimeout,
Clock: mClock,
Model: model,
MaxSteps: 1,
StreamSilenceTimeout: silenceTimeout,
Clock: mClock,
PersistStep: func(_ context.Context, _ PersistedStep) error {
return nil
},
@@ -1071,7 +1204,7 @@ func TestRun_RetriesStartupTimeoutWhenStreamClosesSilently(t *testing.T) {
}()
trap.MustWait(ctx).MustRelease(ctx)
mClock.Advance(startupTimeout).MustWait(ctx)
mClock.Advance(silenceTimeout).MustWait(ctx)
trap.MustWait(ctx).MustRelease(ctx)
require.NoError(t, awaitRunResult(ctx, t, done))
@@ -1087,9 +1220,9 @@ func TestRun_RetriesStartupTimeoutWhenStreamClosesSilently(t *testing.T) {
)
select {
case cause := <-attemptCause:
require.ErrorIs(t, cause, errStartupTimeout)
require.ErrorIs(t, cause, errStreamSilenceTimeout)
case <-ctx.Done():
t.Fatal("timed out waiting for startup timeout cause")
t.Fatal("timed out waiting for silence timeout cause")
}
}