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

@@ -329,8 +329,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 +355,20 @@ export class HttpStream extends Stream implements SqlOwner {
         });
     }
 
+    #fetchWithRetry(request: Request, retryCount = 3): Promise<Response> {
+        try {
+            const fetch = this.#fetch;
+            return 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 +430,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");