fix(opencode): retry stalled SSE streams (#29837)
This commit is contained in:
@@ -48,7 +48,7 @@ function wrapSSE(res: Response, ms: number, ctl: AbortController) {
|
|||||||
async pull(ctrl) {
|
async pull(ctrl) {
|
||||||
const part = await new Promise<Awaited<ReturnType<typeof reader.read>>>((resolve, reject) => {
|
const part = await new Promise<Awaited<ReturnType<typeof reader.read>>>((resolve, reject) => {
|
||||||
const id = setTimeout(() => {
|
const id = setTimeout(() => {
|
||||||
const err = new Error("SSE read timed out")
|
const err = new ProviderError.ResponseStreamError("SSE read timed out")
|
||||||
ctl.abort(err)
|
ctl.abort(err)
|
||||||
void reader.cancel(err)
|
void reader.cancel(err)
|
||||||
reject(err)
|
reject(err)
|
||||||
|
|||||||
@@ -10,6 +10,7 @@ import { testProviderConfig } from "../lib/test-provider"
|
|||||||
import { Env } from "@/env"
|
import { Env } from "@/env"
|
||||||
import { Plugin } from "@/plugin"
|
import { Plugin } from "@/plugin"
|
||||||
import { Provider } from "@/provider/provider"
|
import { Provider } from "@/provider/provider"
|
||||||
|
import { ProviderError } from "@/provider/error"
|
||||||
import { ModelID, ProviderID } from "@/provider/schema"
|
import { ModelID, ProviderID } from "@/provider/schema"
|
||||||
|
|
||||||
afterEach(async () => {
|
afterEach(async () => {
|
||||||
@@ -58,6 +59,35 @@ it.live("headerTimeout does not abort delayed SSE body after headers arrive", ()
|
|||||||
),
|
),
|
||||||
)
|
)
|
||||||
|
|
||||||
|
it.live("chunkTimeout raises a response stream error when SSE body stalls", () =>
|
||||||
|
provideTmpdirServer(
|
||||||
|
({ llm }) =>
|
||||||
|
Effect.gen(function* () {
|
||||||
|
yield* llm.push(reply().wait(Bun.sleep(250)).text("late").stop())
|
||||||
|
|
||||||
|
const provider = yield* Provider.Service
|
||||||
|
const model = yield* provider.getModel(ProviderID.make("test"), ModelID.make("test-model"))
|
||||||
|
const result = streamText({
|
||||||
|
model: yield* provider.getLanguage(model),
|
||||||
|
onError() {},
|
||||||
|
messages: [{ role: "user", content: "hello" }],
|
||||||
|
})
|
||||||
|
|
||||||
|
const error = yield* Effect.promise(async () => {
|
||||||
|
try {
|
||||||
|
for await (const part of result.fullStream) {
|
||||||
|
if (part.type === "error") return part.error
|
||||||
|
}
|
||||||
|
} catch (error) {
|
||||||
|
return error
|
||||||
|
}
|
||||||
|
})
|
||||||
|
expect(error).toBeInstanceOf(ProviderError.ResponseStreamError)
|
||||||
|
}),
|
||||||
|
{ config: (url) => providerConfig(url, { chunkTimeout: 50 }) },
|
||||||
|
),
|
||||||
|
)
|
||||||
|
|
||||||
it.live("headerTimeout aborts when response headers do not arrive", () =>
|
it.live("headerTimeout aborts when response headers do not arrive", () =>
|
||||||
Effect.gen(function* () {
|
Effect.gen(function* () {
|
||||||
const server = yield* Effect.acquireRelease(
|
const server = yield* Effect.acquireRelease(
|
||||||
|
|||||||
Reference in New Issue
Block a user