[AzureStorageFS] more transient errors

hail/python/hailtop/utils/utils.py

@@ -551,6 +551,17 @@ async def bounded_gather2(
 if os.environ.get('HAIL_DONT_RETRY_500') == '1':
     RETRYABLE_HTTP_STATUS_CODES.remove(500)
 
+RETRYABLE_ERRNOS = {
+    # these should match (where an equivalent exists) nettyRetryableErrorNumbers in
+    # is/hail/services/package.scala
+    errno.ETIMEDOUT,
+    errno.ECONNREFUSED,
+    errno.EHOSTUNREACH,
+    errno.ECONNRESET,
+    errno.ENETUNREACH,
+    errno.EPIPE,
+}
+
 
 class TransientError(Exception):
     pass
@@ -656,15 +667,7 @@ def is_transient_error(e):
     if (isinstance(e, aiohttp.ClientPayloadError)
             and e.args[0] == "Response payload is not completed"):
         return True
-    if (isinstance(e, OSError)
-            and e.errno in (errno.ETIMEDOUT,
-                            errno.ECONNREFUSED,
-                            errno.EHOSTUNREACH,
-                            errno.ECONNRESET,
-                            errno.ENETUNREACH,
-                            errno.EPIPE,
-                            errno.ETIMEDOUT
-                            )):
+    if isinstance(e, OSError) and e.errno in RETRYABLE_ERRNOS:
         return True
     if isinstance(e, aiohttp.ClientOSError):
         # aiohttp/client_reqrep.py wraps all OSError instances with a ClientOSError

hail/src/main/scala/is/hail/services/NettyProxy.scala

@@ -0,0 +1,21 @@
+package is.hail.services
+
+import io.netty.channel.unix.Errors  // cannot be in package.scala because is.hail.io shadows top-level io
+import io.netty.channel.unix.Errors.NativeIoException  // cannot be in package.scala because is.hail.io shadows top-level io
+
+object NettyProxy {
+  val ERRNO_ENOENT_NEGATIVE = io.netty.channel.unix.Errors.ERRNO_ENOENT_NEGATIVE
+  val ERRNO_ENOTCONN_NEGATIVE = io.netty.channel.unix.Errors.ERRNO_ENOTCONN_NEGATIVE
+  val ERRNO_EBADF_NEGATIVE = io.netty.channel.unix.Errors.ERRNO_EBADF_NEGATIVE
+  val ERRNO_EPIPE_NEGATIVE = io.netty.channel.unix.Errors.ERRNO_EPIPE_NEGATIVE
+  val ERRNO_ECONNRESET_NEGATIVE = io.netty.channel.unix.Errors.ERRNO_ECONNRESET_NEGATIVE
+  val ERRNO_EAGAIN_NEGATIVE = io.netty.channel.unix.Errors.ERRNO_EAGAIN_NEGATIVE
+  val ERRNO_EWOULDBLOCK_NEGATIVE = io.netty.channel.unix.Errors.ERRNO_EWOULDBLOCK_NEGATIVE
+  val ERRNO_EINPROGRESS_NEGATIVE = io.netty.channel.unix.Errors.ERRNO_EINPROGRESS_NEGATIVE
+  val ERROR_ECONNREFUSED_NEGATIVE = io.netty.channel.unix.Errors.ERROR_ECONNREFUSED_NEGATIVE
+  val ERROR_EISCONN_NEGATIVE = io.netty.channel.unix.Errors.ERROR_EISCONN_NEGATIVE
+  val ERROR_EALREADY_NEGATIVE = io.netty.channel.unix.Errors.ERROR_EALREADY_NEGATIVE
+  val ERROR_ENETUNREACH_NEGATIVE = io.netty.channel.unix.Errors.ERROR_ENETUNREACH_NEGATIVE
+
+  type NativeIoException = io.netty.channel.unix.Errors.NativeIoException
+}

hail/src/main/scala/is/hail/services/package.scala

@@ -56,6 +56,15 @@ package object services {
     }
   }
 
+
+  val nettyRetryableErrorNumbers = Set(
+    // these should match (where an equivalent exists) RETRYABLE_ERRNOS in hailtop/utils/utils.py
+    NettyProxy.ERRNO_EPIPE_NEGATIVE,
+    NettyProxy.ERRNO_ECONNRESET_NEGATIVE,
+    NettyProxy.ERROR_ECONNREFUSED_NEGATIVE,
+    NettyProxy.ERROR_ENETUNREACH_NEGATIVE
+  )
+
   def isTransientError(_e: Throwable): Boolean = {
     // ReactiveException is package private inside reactore.core.Exception so we cannot access
     // it directly for an isInstance check. AFAICT, this is the only way to check if we received
@@ -103,6 +112,17 @@ package object services {
         // is.hail.io.fs.FSSeekableInputStream.read(FS.scala:141)
         // ...
         e.getMessage.contains("Timeout on blocking read")
+      case e: NettyProxy.NativeIoException =>
+        // NativeIoException is a subclass of IOException; therefore this case must appear before
+        // the IOException case
+        //
+        // expectedErr appears to be the additive inverse of the errno returned by Linux?
+        //
+        // https://github.com/netty/netty/blob/24a0ac36ea91d1aee647d738f879ac873892d829/transport-native-unix-common/src/main/java/io/netty/channel/unix/Errors.java#L49
+        (nettyRetryableErrorNumbers.contains(e.expectedErr) ||
+          // io.netty.channel.unix.Errors$NativeIoException: readAddress(..) failed: Connection reset by peer
+          e.getMessage.contains("Connection reset by peer")
+        )
       case e @ (_: SSLException | _: StorageException | _: IOException) =>
         val cause = e.getCause
         cause != null && isTransientError(cause)