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.

Author: penberg

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: number = 3, backoff: number = 100): Promise<Response> {
+        const fetch = this.#fetch;
+        return fetch(request).catch((error) => {
+            if (isRetryableError(error)) {
+                if (retryCount > 0) {
+                    return new Promise((resolve) => setTimeout(resolve, backoff)).then(() => {
+                        return this.#fetchWithRetry(request, retryCount - 1, backoff * 2);
+                    });
+                }
+            }
+            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");