Handle throttling in FileDescriptor implementations

.github/workflows/ci.yml

@@ -41,7 +41,7 @@ jobs:
           git diff --exit-code
 
       - name: Run tests
-        run: sbt -Dsbt.color=always "++${{ matrix.scala }} test"
+        run: sbt -Dsbt.color=always "++${{ matrix.scala }} testFull"
 
       - name: Check mdoc for uncommitted changes
         if: matrix.scala == '2.13'

README.md

@@ -338,6 +338,21 @@ The `CredentialStore` object serves as an endpoint for the retrieval of AWS cred
 
 The `S3Bucket` class wraps an instance of `S3AsyncClient` (from AWS SDK for Java) and exposes a higher level interface for pushing and pulling files to and from a bucket.
 
+It reads the following optional keys from the typesafe configuration, leaving the corresponding behavior of the underlying client untouched when a key is unset:
+
+| Key | Default | Description |
+| --- | --- | --- |
+| `aws.s3.region` | the client's default region | The location constraint with which to create the bucket, when it doesn't exist yet. |
+| `aws.s3.max-connections` | computed by the client from its target throughput | The maximum number of S3 connections that should be established during a transfer. |
+| `aws.s3.max-error-retry` | the client's default retry configuration | The maximum number of retry attempts performed by the underlying client for failed retryable requests. |
+| `aws.s3.retry.on-slow-down` | `true` | Whether to retry a request that S3 throttled, identified whether it is reported as a service error or as a client-side error. |
+| `aws.s3.retry.max-retries` | `2` | The maximum number of times to retry an operation, on top of its first attempt. |
+| `aws.s3.retry.base-backoff` | `3 seconds` | The duration to wait before the first retry, doubling on every subsequent one. |
+| `aws.s3.retry.max-backoff` | `30 seconds` | The duration with which to cap the growth of the waiting duration. |
+| `aws.s3.retry.jitter` | `1.0` | The fraction of each waiting duration to randomize, from `0.0` for no jitter to `1.0` for a duration uniformly distributed between zero and the full waiting duration. |
+
+Failed operations that are worth retrying are retried waiting for an exponentially growing duration, jittered so that concurrent callers don't retry in lockstep. Note that the `aws.s3.retry` keys configure these retries, performed by `S3Bucket` itself, while `aws.s3.max-error-retry` configures the ones performed by the underlying client.
+
 ### SerializableAWSCredentials
 
 The `SerializableAWSCredentials` class provides a serializable container for AWS credentials, extending the `AwsCredentials` class (from AWS SDK for Java).
@@ -627,6 +642,17 @@ libraryDependencies += "com.kevel" %% "apso-gcp" % "0.27.4"
 
 The `GCSBucket` class wraps an instance of `Storage` (from GCS SDK for Java) and exposes a higher level interface for pushing and pulling files to and from a bucket.
 
+It reads the following optional keys from the typesafe configuration:
+
+| Key | Default | Description |
+| --- | --- | --- |
+| `gcp.storage.retry.max-retries` | `2` | The maximum number of times to retry an operation, on top of its first attempt. |
+| `gcp.storage.retry.base-backoff` | `3 seconds` | The duration to wait before the first retry, doubling on every subsequent one. |
+| `gcp.storage.retry.max-backoff` | `30 seconds` | The duration with which to cap the growth of the waiting duration. |
+| `gcp.storage.retry.jitter` | `1.0` | The fraction of each waiting duration to randomize, from `0.0` for no jitter to `1.0` for a duration uniformly distributed between zero and the full waiting duration. |
+
+These keys configure the retries performed by `GCSBucket` itself, which are independent from the ones performed by the underlying client. An operation is retried whenever the client reports its failure as retryable, which covers the requests GCS throttles.
+
 ## Encryption
 
 Apso provides some simple utility classes to deal with encryption and decryption of data, and methods that ease the

build.sbt

@@ -112,7 +112,8 @@ lazy val aws = module(project, "aws")
       AwsSdkS3,
       AwsSdkS3Transfer,
       ScalaLogging,
-      TypesafeConfig
+      TypesafeConfig,
+      Specs2_4Core % Test
     )
   )
 

docs/README.md

@@ -321,6 +321,21 @@ The `CredentialStore` object serves as an endpoint for the retrieval of AWS cred
 
 The `S3Bucket` class wraps an instance of `S3AsyncClient` (from AWS SDK for Java) and exposes a higher level interface for pushing and pulling files to and from a bucket.
 
+It reads the following optional keys from the typesafe configuration, leaving the corresponding behavior of the underlying client untouched when a key is unset:
+
+| Key | Default | Description |
+| --- | --- | --- |
+| `aws.s3.region` | the client's default region | The location constraint with which to create the bucket, when it doesn't exist yet. |
+| `aws.s3.max-connections` | computed by the client from its target throughput | The maximum number of S3 connections that should be established during a transfer. |
+| `aws.s3.max-error-retry` | the client's default retry configuration | The maximum number of retry attempts performed by the underlying client for failed retryable requests. |
+| `aws.s3.retry.on-slow-down` | `true` | Whether to retry a request that S3 throttled, identified whether it is reported as a service error or as a client-side error. |
+| `aws.s3.retry.max-retries` | `2` | The maximum number of times to retry an operation, on top of its first attempt. |
+| `aws.s3.retry.base-backoff` | `3 seconds` | The duration to wait before the first retry, doubling on every subsequent one. |
+| `aws.s3.retry.max-backoff` | `30 seconds` | The duration with which to cap the growth of the waiting duration. |
+| `aws.s3.retry.jitter` | `1.0` | The fraction of each waiting duration to randomize, from `0.0` for no jitter to `1.0` for a duration uniformly distributed between zero and the full waiting duration. |
+
+Failed operations that are worth retrying are retried waiting for an exponentially growing duration, jittered so that concurrent callers don't retry in lockstep. Note that the `aws.s3.retry` keys configure these retries, performed by `S3Bucket` itself, while `aws.s3.max-error-retry` configures the ones performed by the underlying client.
+
 ### SerializableAWSCredentials
 
 The `SerializableAWSCredentials` class provides a serializable container for AWS credentials, extending the `AwsCredentials` class (from AWS SDK for Java).
@@ -490,6 +505,17 @@ libraryDependencies += "com.kevel" %% "apso-gcp" % "@VERSION@"
 
 The `GCSBucket` class wraps an instance of `Storage` (from GCS SDK for Java) and exposes a higher level interface for pushing and pulling files to and from a bucket.
 
+It reads the following optional keys from the typesafe configuration:
+
+| Key | Default | Description |
+| --- | --- | --- |
+| `gcp.storage.retry.max-retries` | `2` | The maximum number of times to retry an operation, on top of its first attempt. |
+| `gcp.storage.retry.base-backoff` | `3 seconds` | The duration to wait before the first retry, doubling on every subsequent one. |
+| `gcp.storage.retry.max-backoff` | `30 seconds` | The duration with which to cap the growth of the waiting duration. |
+| `gcp.storage.retry.jitter` | `1.0` | The fraction of each waiting duration to randomize, from `0.0` for no jitter to `1.0` for a duration uniformly distributed between zero and the full waiting duration. |
+
+These keys configure the retries performed by `GCSBucket` itself, which are independent from the ones performed by the underlying client. An operation is retried whenever the client reports its failure as retryable, which covers the requests GCS throttles.
+
 ## Encryption
 
 Apso provides some simple utility classes to deal with encryption and decryption of data, and methods that ease the

modules/aws/src/main/scala/com/kevel/apso/aws/S3Bucket.scala

@@ -4,21 +4,26 @@ import java.io.*
 import java.util.concurrent.atomic.AtomicInteger
 import java.util.concurrent.{CompletableFuture, CompletionException, LinkedBlockingQueue, ThreadPoolExecutor, TimeUnit}
 
+import scala.concurrent.duration.*
 import scala.jdk.CollectionConverters.*
-import scala.util.{Failure, Success, Try, Using}
+import scala.util.{Try, Using}
 
 import com.typesafe.config.ConfigFactory
 import com.typesafe.scalalogging.LazyLogging
 import software.amazon.awssdk.auth.credentials.AwsCredentialsProvider
 import software.amazon.awssdk.core.ResponseInputStream
 import software.amazon.awssdk.core.async.{AsyncRequestBody, AsyncResponseTransformer}
 import software.amazon.awssdk.core.exception.{SdkClientException, SdkException}
+import software.amazon.awssdk.core.retry.RetryUtils
 import software.amazon.awssdk.regions
 import software.amazon.awssdk.services.s3.S3AsyncClient
 import software.amazon.awssdk.services.s3.crt.S3CrtRetryConfiguration
 import software.amazon.awssdk.services.s3.model.*
 import software.amazon.awssdk.transfer.s3.{S3TransferManager, model}
 
+import com.kevel.apso.Retry
+import com.kevel.apso.aws.S3Bucket.isSlowDown
+
 /** A representation of an Amazon's S3 bucket. This class wraps an `S3AsyncClient` and provides a higher level interface
   * for pushing and pulling files to and from a bucket.
   *
@@ -37,10 +42,23 @@ class S3Bucket(
   private[this] lazy val config = ConfigFactory.load()
 
   private[this] lazy val configPrefix = "aws.s3"
+  private[this] lazy val retryPrefix = configPrefix + ".retry"
+
   private[this] lazy val region = Try(config.getString(configPrefix + ".region"))
   private[this] lazy val maxConnections = Try(config.getInt(configPrefix + ".max-connections"))
   private[this] lazy val maxErrorRetry = Try(config.getInt(configPrefix + ".max-error-retry"))
 
+  private[this] lazy val retryOnSlowDown =
+    Try(config.getBoolean(retryPrefix + ".on-slow-down")).getOrElse(true)
+  private[this] lazy val retryMaxRetries =
+    Try(config.getInt(retryPrefix + ".max-retries")).getOrElse(S3Bucket.DefaultMaxRetries)
+  private[this] lazy val retryBaseBackOff =
+    Try(config.getDuration(retryPrefix + ".base-backoff").toMillis.millis).getOrElse(S3Bucket.DefaultBaseBackOff)
+  private[this] lazy val retryMaxBackOff =
+    Try(config.getDuration(retryPrefix + ".max-backoff").toMillis.millis).getOrElse(S3Bucket.DefaultMaxBackOff)
+  private[this] lazy val retryJitter =
+    Try(config.getDouble(retryPrefix + ".jitter")).getOrElse(S3Bucket.DefaultJitter)
+
   @transient private[this] lazy val defaultExecutor = {
     val maxPoolSize = 100
     val threadCount = new AtomicInteger(0)
@@ -169,14 +187,15 @@ class S3Bucket(
     * @return
     *   a list of objects in a bucket matching a given prefix.
     */
-  def getObjectsWithMatchingPrefix(prefix: String, includeDirectories: Boolean = false): Iterator[S3Object] = retry {
-    logger.info(s"Finding files matching prefix '$prefix'...")
+  def getObjectsWithMatchingPrefix(prefix: String, includeDirectories: Boolean = false): Iterator[S3Object] =
+    retry {
+      logger.info(s"Finding files matching prefix '$prefix'...")
 
-    val req = ListObjectsV2Request.builder.bucket(bucketName).prefix(sanitizeKey(prefix)).build
-    val objects = listObjectsV2Iterator(req).flatMap(_.contents.asScala)
+      val req = ListObjectsV2Request.builder.bucket(bucketName).prefix(sanitizeKey(prefix)).build
+      val objects = listObjectsV2Iterator(req).flatMap(_.contents.asScala)
 
-    if (includeDirectories) objects else objects.filterNot(_.key.endsWith("/"))
-  }.getOrElse(Iterator.empty)
+      if (includeDirectories) objects else objects.filterNot(_.key.endsWith("/"))
+    }.getOrElse(Iterator.empty)
 
   // FIXME: If the root directory/prefix was created by the `mkdirs` method (where we create an object with 0 bytes)
   //        that root directory will be present in the results. Evaluate if we should filter it out since it does not
@@ -413,10 +432,33 @@ class S3Bucket(
     stream
   }
 
+  private[aws] def retry[T](f: => T): Option[T] =
+    Retry
+      .exponentialBackOff(
+        maxRetries = retryMaxRetries,
+        base = retryBaseBackOff,
+        max = Some(retryMaxBackOff),
+        jitter = retryJitter,
+        // `handler` is partial, so anything it doesn't classify is treated as worth retrying rather than thrown.
+        retryWhen = ex => !handler.applyOrElse(ex, (_: Throwable) => false),
+        // The failure itself is already logged by `handler`, so it isn't logged again here.
+        onRetry = (_, delay, remaining) =>
+          logger.warn(s"Error during S3 operation. Retrying in ${delay.toMillis}ms ($remaining more times)"),
+        onMaxRetriesReached = _ => logger.error("Max retries reached. Aborting S3 operation")
+      )(f)
+      .toOption
+
   private def log(isError: Boolean, message: String, cause: Throwable): Unit =
     if (isError) logger.error(message, cause) else logger.warn(message, cause)
 
   private[this] def handler: PartialFunction[Throwable, Boolean] = {
+    // Matched ahead of the shape-specific cases below, since a slow-down is reported both as a service error and as a
+    // client-side error, depending on the client in use. The CRT-based client reports it as the latter with `retryable`
+    // set to false, which is why the cases below can't be relied on to retry throttled requests.
+    case ex: SdkException if isSlowDown(ex) =>
+      log(!retryOnSlowDown, s"S3 slow down: ${ex.getMessage}", ex)
+      !retryOnSlowDown
+
     case ex: S3Exception =>
       ex.statusCode() match {
         case 404 =>
@@ -428,11 +470,12 @@ class S3Bucket(
         case _ =>
           logger.warn(
             s"""|S3 service error: ${ex.getMessage}. Extended request id: ${ex.requestId}
-                      |Message: ${ex.getMessage}""".stripMargin,
+                |Message: ${ex.getMessage}""".stripMargin,
             ex
           )
           false
       }
+
     case ex: SdkClientException =>
       log(!ex.retryable, s"Client Exception: ${ex.getMessage}", ex)
       !ex.retryable
@@ -450,25 +493,33 @@ class S3Bucket(
       false
   }
 
-  private[this] def retry[T](f: => T, tries: Int = 3, sleepTime: Int = 5000): Option[T] =
-    if (tries == 0) {
-      logger.error("Max retries reached. Aborting S3 operation")
-      None
-    } else
-      Try(f) match {
-        case Success(res)              => Some(res)
-        case Failure(e) if !handler(e) =>
-          if (tries > 1) {
-            logger.warn(s"Error during S3 operation. Retrying in ${sleepTime}ms (${tries - 1} more times)")
-            Thread.sleep(sleepTime)
-          }
-          retry(f, tries - 1, sleepTime)
-
-        case _ => None
-      }
-
   override def equals(obj: Any): Boolean = obj match {
     case b: S3Bucket => b.bucketName == bucketName
     case _           => false
   }
 }
+
+object S3Bucket extends LazyLogging {
+  private[aws] val DefaultMaxRetries = 2
+  private[aws] val DefaultBaseBackOff = 3.seconds
+  private[aws] val DefaultMaxBackOff = 30.seconds
+  private[aws] val DefaultJitter = 1d
+
+  /** The error string with which the CRT-based S3 client reports a throttled request.
+    *
+    * The CRT signals throttling through its `AWS_ERROR_S3_SLOW_DOWN` error, which reaches the SDK as an
+    * `SdkClientException` whose message embeds only the rendered error string, and not the numeric error code.
+    */
+  private[aws] val SlowDownErrorMessage = "Response code indicates throttling"
+
+  /** Returns whether the given exception reports S3 having throttled the request.
+    *
+    * A throttled request surfaces either as a service error, which S3 reports with a `503` status code, or as a
+    * client-side error, which is how the CRT-based client reports it.
+    */
+  private[aws] def isSlowDown(ex: SdkException): Boolean =
+    RetryUtils.isThrottlingException(ex) || (ex match {
+      case ex: SdkClientException => Option(ex.getMessage).exists(_.contains(SlowDownErrorMessage))
+      case _                      => false
+    })
+}

modules/aws/src/test/scala/com/kevel/apso/aws/S3BucketSpec.scala

@@ -0,0 +1,59 @@
+package com.kevel.apso.aws
+
+import org.specs2.mutable.Specification
+import software.amazon.awssdk.awscore.exception.AwsErrorDetails
+import software.amazon.awssdk.core.exception.SdkClientException
+import software.amazon.awssdk.services.s3.model.S3Exception
+
+class S3BucketSpec extends Specification {
+
+  // The message the CRT-based S3 client builds for a failed request, as seen in production.
+  private def clientError(errorString: String) =
+    SdkClientException.create(s"Failed to send the request: $errorString")
+
+  private def serviceError(statusCode: Int, errorCode: String) =
+    S3Exception
+      .builder()
+      .statusCode(statusCode)
+      .awsErrorDetails(AwsErrorDetails.builder().errorCode(errorCode).serviceName("S3").build())
+      .message("Boom")
+      .build()
+
+  "An S3Bucket" should {
+
+    "recognize a slow down" in {
+
+      "reported as a service error" in {
+        S3Bucket.isSlowDown(serviceError(503, "SlowDown")) must beTrue
+      }
+
+      "reported as a service error with a 429 status code" in {
+        S3Bucket.isSlowDown(serviceError(429, "TooManyRequestsException")) must beTrue
+      }
+
+      "reported by the CRT client as a client-side error" in {
+        S3Bucket.isSlowDown(clientError(S3Bucket.SlowDownErrorMessage)) must beTrue
+      }
+    }
+
+    "not recognize as a slow down" in {
+
+      // A 503 alone doesn't imply throttling, so it must keep being handled as a plain service error.
+      "a service error with a 503 status code but no throttling error code" in {
+        S3Bucket.isSlowDown(serviceError(503, "ServiceUnavailable")) must beFalse
+      }
+
+      "a service error reporting another failure" in {
+        S3Bucket.isSlowDown(serviceError(404, "NoSuchKey")) must beFalse
+      }
+
+      "a client-side error reporting another failure" in {
+        S3Bucket.isSlowDown(clientError("Socket closed")) must beFalse
+      }
+
+      "a client-side error without a message" in {
+        S3Bucket.isSlowDown(SdkClientException.builder().build()) must beFalse
+      }
+    }
+  }
+}