Add retries to Data.read

distribution/lib/Standard/Base/0.0.0-dev/src/Internal/Data_Read_Helpers.enso

@@ -5,11 +5,13 @@ import project.Data.Text.Text
 import project.Data.Vector.Vector
 import project.Enso_Cloud.Data_Link.Data_Link
 import project.Enso_Cloud.Data_Link_Helpers
+import project.Error.Error
 import project.Errors.Deprecated.Deprecated
 import project.Errors.Problem_Behavior.Problem_Behavior
 import project.Metadata.Display
 import project.Metadata.Widget
 import project.Network.HTTP.HTTP
+import project.Network.HTTP.HTTP_Error.HTTP_Error
 import project.Network.HTTP.HTTP_Method.HTTP_Method
 import project.Network.URI.URI
 import project.Warning.Warning
@@ -20,6 +22,8 @@ from project.Metadata.Choice import Option
 from project.Metadata.Widget import Single_Choice
 from project.System.File_Format import Auto_Detect, format_types
 
+polyglot java import java.lang.Thread
+
 ## PRIVATE
 looks_like_uri path:Text -> Boolean =
     (path.starts_with "http://" Case_Sensitivity.Insensitive) || (path.starts_with "https://" Case_Sensitivity.Insensitive)
@@ -28,8 +32,19 @@ looks_like_uri path:Text -> Boolean =
    A common implementation for fetching a resource and decoding it,
    following encountered data links.
 fetch_following_data_links (uri:URI) (method:HTTP_Method = HTTP_Method.Get) (headers:Vector = []) format =
-    response = HTTP.fetch uri method headers
-    decode_http_response_following_data_links response format
+    fetch_and_decode =
+        response = HTTP.fetch uri method headers
+        decode_http_response_following_data_links response format
+
+    error_handler attempt =
+        caught_error ->
+            exponential_backoff = [100, 200, 400]
+            if method != HTTP_Method.Get || attempt >= exponential_backoff.length then Error.throw caught_error else
+                sleep_time = exponential_backoff . get attempt
+                Thread.sleep sleep_time
+                fetch_and_decode . catch HTTP_Error (error_handler (attempt + 1))
+
+    fetch_and_decode . catch HTTP_Error (error_handler 0)
 
 ## PRIVATE
    Decodes a HTTP response, handling data link access.

distribution/lib/Standard/Base/0.0.0-dev/src/Runtime/Thread.enso

@@ -2,6 +2,7 @@
    Internal threading utilities used for working with threads.
 
 import project.Any.Any
+import project.Data.Numbers.Integer
 
 ## PRIVATE
    ADVANCED

test/Base_Tests/src/Network/Http_Spec.enso

@@ -591,7 +591,7 @@ add_specs suite_builder =
 
         ## Checking this error partially as a warning - I spent a lot of time debugging why I'm getting such an error.
            Apparently it happens when the httpbin server was crashing without sending any response.
-        group_builder.specify "should be able to handle server crash resulting in no response" pending=pending_has_url <| Test.with_retries <|
+        group_builder.specify "should be able to handle server crash resulting in no response" pending=pending_has_url <|
             err = Data.fetch (base_url_with_slash+"crash")
             err.should_fail_with Request_Error
             err.catch.error_type . should_equal "java.io.IOException"
@@ -600,10 +600,20 @@ add_specs suite_builder =
                I think it may be worth adding, because it may be really quite confusing for end users who get that kind of error.
             err.catch.message . should_equal "HTTP/1.1 header parser received no bytes"
 
-        group_builder.specify "should be able to handle IO errors" pending="TODO: Currently I was unable to figure out a way to test such errors" <|
-            # how to trigger this error???
-            err = Data.fetch "TODO"
+        group_builder.specify "should be able to handle occasional server crashes and retry" pending=pending_has_url <|
+            r1 = Data.fetch (base_url_with_slash+"crash?success_every=2")
+            r1.should_succeed
+            r1.should_be_a Response
+
+        group_builder.specify "should be able to handle server crash that closes stream abruptly" pending=pending_has_url <|
+            err = Data.fetch (base_url_with_slash+"crash?type=stream")
             err.should_fail_with HTTP_Error
+            err.catch.message . should_equal "An IO error has occurred: java.io.IOException: closed"
+
+        group_builder.specify "should be able to handle occasional abrupt stream closures and retry" pending=pending_has_url <|
+            r1 = Data.fetch (base_url_with_slash+"crash?type=stream&success_every=2")
+            r1.should_succeed
+            r1.should_be_a Response
 
     suite_builder.group "Http Auth" group_builder->
         group_builder.specify "should support Basic user+password authentication" pending=pending_has_url <| Test.with_retries <|

tools/http-test-helper/src/main/java/org/enso/shttp/test_helpers/CrashingTestHandler.java

@@ -2,13 +2,83 @@
 
 import com.sun.net.httpserver.HttpExchange;
 import java.io.IOException;
+import java.io.OutputStream;
+import java.net.URI;
+import org.apache.http.client.utils.URIBuilder;
 import org.enso.shttp.SimpleHttpHandler;
 
 public class CrashingTestHandler extends SimpleHttpHandler {
+  private int requests = 0;
+
   @Override
   protected void doHandle(HttpExchange exchange) throws IOException {
-    // This exception will be logged by SimpleHttpHandler, but that's OK - let's know that this
+    // Exceptions will be logged by SimpleHttpHandler, but that's OK - let's know that this
     // crash is happening.
-    throw new RuntimeException("This handler crashes on purpose.");
+
+    URI uri = exchange.getRequestURI();
+    URIBuilder builder = new URIBuilder(uri);
+    final String successEveryParam = "success_every";
+    final String crashTypeParam = "type";
+
+    int successEvery = 0;
+    CrashType crashType = CrashType.RUNTIME;
+
+    for (var queryPair : builder.getQueryParams()) {
+      if (queryPair.getName().equals(successEveryParam)) {
+        successEvery = Integer.decode(queryPair.getValue());
+      } else if (queryPair.getName().equals(crashTypeParam)) {
+        crashType =
+            switch (queryPair.getValue()) {
+              case "stream" -> CrashType.STREAM;
+              default -> CrashType.RUNTIME;
+            };
+      }
+    }
+    if (successEvery == 0) {
+      // Reset counter
+      requests = 0;
+    }
+
+    boolean shouldSucceed = successEvery == (requests + 1);
+
+    switch (crashType) {
+      case RUNTIME:
+        if (shouldSucceed) {
+          // return OK, reset
+          requests = 0;
+          exchange.sendResponseHeaders(200, -1);
+          exchange.close();
+          break;
+        } else {
+          requests += 1;
+          throw new RuntimeException("This handler crashes on purpose.");
+        }
+
+      case STREAM:
+        byte[] responseData = "Crash and Burn".getBytes();
+        exchange.sendResponseHeaders(200, responseData.length);
+        try {
+          if (shouldSucceed) {
+            requests = 0;
+            // return OK, reset
+            try (OutputStream os = exchange.getResponseBody()) {
+              os.write(responseData, 0, responseData.length);
+            }
+          } else {
+            requests += 1;
+            try (OutputStream os = exchange.getResponseBody()) {
+              os.write(responseData, 0, responseData.length / 2);
+            }
+          }
+        } finally {
+          exchange.close();
+        }
+        break;
+    }
+  }
+
+  enum CrashType {
+    RUNTIME, // Simulate an internal server crash
+    STREAM // Simulate a crash by abruptly closing response's body stream
   }
 }