Fix flush() to retry fetch() on transient errors

This changes flush() to retry the fetch() call on transient errors such
as EPIPE, ECONNREFUSED, and ECONNRESET.

src/http/stream.ts

@@ -31,6 +31,8 @@ import { CursorReqBody as json_CursorReqBody } from "./json_encode.js";
 import { CursorReqBody as protobuf_CursorReqBody } from "./protobuf_encode.js";
 import { PipelineRespBody as json_PipelineRespBody } from "./json_decode.js";
 import { PipelineRespBody as protobuf_PipelineRespBody } from "./protobuf_decode.js";
+import { FetchCursorReq } from "../ws/proto.js";
+import { FetchError } from "node-fetch";
 
 type QueueEntry = PipelineEntry | CursorEntry;
 
@@ -329,8 +331,7 @@ export class HttpStream extends Stream implements SqlOwner {
         let promise;
         try {
             const request = createRequest();
-            const fetch = this.#fetch;
-            promise = fetch(request);
+            promise = this.#fetchWithRetry(request);
         } catch (error) {
             promise = Promise.reject(error);
         }
@@ -356,6 +357,19 @@ export class HttpStream extends Stream implements SqlOwner {
         });
     }
 
+    #fetchWithRetry(request: Request, retryCount = 3): Promise<Response> {
+        try {
+            return this.#fetch(request);
+        } catch (error: any) {
+            if (isRetryableError(error)) {
+                if (retryCount > 0) {
+                    return this.#fetchWithRetry(request, retryCount - 1);
+                }
+            }
+            throw error;
+        }
+    }
+
     #createPipelineRequest(pipeline: Array<PipelineEntry>, endpoint: Endpoint): Request {
         return this.#createRequest<proto.PipelineReqBody>(
             new URL(endpoint.pipelinePath, this.#baseUrl),
@@ -417,6 +431,15 @@ export class HttpStream extends Stream implements SqlOwner {
     }
 }
 
+function isRetryableError(error: any): boolean {
+    if (!error.errno) {
+        return false;
+    }
+    return error.errno === "EPIPE"
+        || error.errno === "ECONNREFUSED"
+        || error.errno === "ECONNRESET";
+}
+
 function handlePipelineResponse(pipeline: Array<PipelineEntry>, respBody: proto.PipelineRespBody): void {
     if (respBody.results.length !== pipeline.length) {
         throw new ProtoError("Server returned unexpected number of pipeline results");