Skip to content

Commit 4d44ee0

Browse files
jincheng-xydtclaude
jincheng-xydt
and
claude
committed
fix(bedrock): retry Converse Streaming on transient connection drops (#87876)
Bedrock Converse Streaming silently drops the connection after ~6 minutes on large prompts, leaving no transcript, no retry, and no fallback model attempt. This adds one automatic retry for transient stream failures. - Wrap client.send() + stream iteration in a retry loop (max 1 retry) - Classify errors with isTransientBedrockStreamError(): - Transient: InternalServerException, ServiceUnavailableException, ModelStreamErrorException, AbortError, socket/network errors - Permanent: ValidationException, ThrottlingException, auth errors - Create a fresh BedrockRuntimeClient on retry for clean TCP connection - Reset output content and usage on retry - Add regression tests for retry behavior and error classification Fixes #87876 Co-Authored-By: Claude <[email protected]>
1 parent 4e2f015 commit 4d44ee0

2 files changed

Lines changed: 329 additions & 59 deletions

File tree

‎extensions/amazon-bedrock/stream.runtime.test.ts‎

Lines changed: 191 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -501,3 +501,194 @@ describe("Bedrock canonical Claude aliases", () => {
501501
},
502502
);
503503
});
504+
505+
describe("Bedrock stream retry (#87876)", () => {
506+
function context() {
507+
return {
508+
messages: [{ role: "user", content: "Reply briefly.", timestamp: 0 }],
509+
} as never;
510+
}
511+
512+
it("retries once on InternalServerException from client.send", async () => {
513+
const send = vi
514+
.spyOn(BedrockRuntimeClient.prototype, "send")
515+
.mockRejectedValueOnce(
516+
Object.assign(new Error("Internal server error"), {
517+
name: "InternalServerException",
518+
$metadata: {},
519+
}),
520+
)
521+
.mockResolvedValueOnce({
522+
$metadata: { httpStatusCode: 200 },
523+
stream: streamEvents([
524+
{ messageStart: { role: ConversationRole.ASSISTANT } },
525+
{ messageStop: { stopReason: "end_turn" } },
526+
]),
527+
} as never);
528+
529+
const stream = streamSimpleBedrock(bedrockModel({}), context());
530+
await stream.result();
531+
532+
expect(send).toHaveBeenCalledTimes(2);
533+
});
534+
535+
it("retries once on ServiceUnavailableException", async () => {
536+
const send = vi
537+
.spyOn(BedrockRuntimeClient.prototype, "send")
538+
.mockRejectedValueOnce(
539+
Object.assign(new Error("Service unavailable"), {
540+
name: "ServiceUnavailableException",
541+
$metadata: {},
542+
}),
543+
)
544+
.mockResolvedValueOnce({
545+
$metadata: { httpStatusCode: 200 },
546+
stream: streamEvents([
547+
{ messageStart: { role: ConversationRole.ASSISTANT } },
548+
{ messageStop: { stopReason: "end_turn" } },
549+
]),
550+
} as never);
551+
552+
const stream = streamSimpleBedrock(bedrockModel({}), context());
553+
await stream.result();
554+
555+
expect(send).toHaveBeenCalledTimes(2);
556+
});
557+
558+
it("does not retry on ValidationException (permanent)", async () => {
559+
const send = vi
560+
.spyOn(BedrockRuntimeClient.prototype, "send")
561+
.mockRejectedValue(
562+
Object.assign(new Error("Validation error"), {
563+
name: "ValidationException",
564+
$metadata: {},
565+
}),
566+
);
567+
568+
const stream = streamSimpleBedrock(bedrockModel({}), context());
569+
const result = await stream.result();
570+
571+
expect(result.errorMessage).toContain("Validation error");
572+
expect(send).toHaveBeenCalledTimes(1);
573+
});
574+
575+
it("does not retry on ThrottlingException (permanent)", async () => {
576+
const send = vi
577+
.spyOn(BedrockRuntimeClient.prototype, "send")
578+
.mockRejectedValue(
579+
Object.assign(new Error("Throttling error"), {
580+
name: "ThrottlingException",
581+
$metadata: {},
582+
}),
583+
);
584+
585+
const stream = streamSimpleBedrock(bedrockModel({}), context());
586+
const result = await stream.result();
587+
588+
expect(result.errorMessage).toContain("Throttling error");
589+
expect(send).toHaveBeenCalledTimes(1);
590+
});
591+
592+
it("does not retry when the caller explicitly aborted (user-initiated)", async () => {
593+
const controller = new AbortController();
594+
controller.abort();
595+
596+
const send = vi.spyOn(BedrockRuntimeClient.prototype, "send");
597+
598+
const stream = streamSimpleBedrock(bedrockModel({}), context(), {
599+
signal: controller.signal,
600+
});
601+
const result = await stream.result();
602+
603+
expect(result.errorMessage).toBeDefined();
604+
expect(send).toHaveBeenCalledTimes(0); // AbortSignal already fired
605+
});
606+
607+
it("retries once on stream ModelStreamErrorException", async () => {
608+
const send = vi
609+
.spyOn(BedrockRuntimeClient.prototype, "send")
610+
.mockResolvedValueOnce({
611+
$metadata: { httpStatusCode: 200 },
612+
stream: streamEvents([
613+
{ messageStart: { role: ConversationRole.ASSISTANT } },
614+
{ modelStreamErrorException: Object.assign(new Error("stream error"), { name: "ModelStreamErrorException" }) },
615+
]),
616+
} as never)
617+
.mockResolvedValueOnce({
618+
$metadata: { httpStatusCode: 200 },
619+
stream: streamEvents([
620+
{ messageStart: { role: ConversationRole.ASSISTANT } },
621+
{ messageStop: { stopReason: "end_turn" } },
622+
]),
623+
} as never);
624+
625+
const stream = streamSimpleBedrock(bedrockModel({}), context());
626+
await stream.result();
627+
628+
expect(send).toHaveBeenCalledTimes(2);
629+
});
630+
});
631+
632+
describe("isTransientBedrockStreamError", () => {
633+
it("returns true for InternalServerException", () => {
634+
const err = Object.assign(new Error("boom"), {
635+
name: "InternalServerException",
636+
$metadata: {},
637+
});
638+
expect(testing.isTransientBedrockStreamError(err)).toBe(true);
639+
});
640+
641+
it("returns true for ServiceUnavailableException", () => {
642+
const err = Object.assign(new Error("down"), {
643+
name: "ServiceUnavailableException",
644+
$metadata: {},
645+
});
646+
expect(testing.isTransientBedrockStreamError(err)).toBe(true);
647+
});
648+
649+
it("returns true for ModelStreamErrorException", () => {
650+
const err = Object.assign(new Error("stream broke"), {
651+
name: "ModelStreamErrorException",
652+
$metadata: {},
653+
});
654+
expect(testing.isTransientBedrockStreamError(err)).toBe(true);
655+
});
656+
657+
it("returns false for ValidationException", () => {
658+
const err = Object.assign(new Error("bad input"), {
659+
name: "ValidationException",
660+
$metadata: {},
661+
});
662+
expect(testing.isTransientBedrockStreamError(err)).toBe(false);
663+
});
664+
665+
it("returns false for ThrottlingException", () => {
666+
const err = Object.assign(new Error("rate limited"), {
667+
name: "ThrottlingException",
668+
$metadata: {},
669+
});
670+
expect(testing.isTransientBedrockStreamError(err)).toBe(false);
671+
});
672+
673+
it("returns false for auth errors", () => {
674+
const err = new Error("Unauthorized: Access denied");
675+
expect(testing.isTransientBedrockStreamError(err)).toBe(false);
676+
});
677+
678+
it("returns true for AbortError", () => {
679+
const err = Object.assign(new Error("The operation was aborted"), {
680+
name: "AbortError",
681+
});
682+
expect(testing.isTransientBedrockStreamError(err)).toBe(true);
683+
});
684+
685+
it("returns true for socket hang-up errors", () => {
686+
const err = new Error("socket hang up");
687+
expect(testing.isTransientBedrockStreamError(err)).toBe(true);
688+
});
689+
690+
it("returns true for ECONNRESET", () => {
691+
const err = new Error("read ECONNRESET");
692+
expect(testing.isTransientBedrockStreamError(err)).toBe(true);
693+
});
694+
});

0 commit comments

Comments
 (0)