diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml
index 22fb2b33..231e9fb2 100644
--- a/.github/workflows/ci.yml
+++ b/.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'
diff --git a/README.md b/README.md
index 445afe36..b5ac4afb 100644
--- a/README.md
+++ b/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
diff --git a/build.sbt b/build.sbt
index c0f889ca..c4b26dea 100644
--- a/build.sbt
+++ b/build.sbt
@@ -112,7 +112,8 @@ lazy val aws = module(project, "aws")
       AwsSdkS3,
       AwsSdkS3Transfer,
       ScalaLogging,
-      TypesafeConfig
+      TypesafeConfig,
+      Specs2_4Core % Test
     )
   )
 
diff --git a/docs/README.md b/docs/README.md
index 6f9cf991..365bc17a 100644
--- a/docs/README.md
+++ b/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
diff --git a/modules/aws/src/main/scala/com/kevel/apso/aws/S3Bucket.scala b/modules/aws/src/main/scala/com/kevel/apso/aws/S3Bucket.scala
index 54dafbc1..d1afc850 100644
--- a/modules/aws/src/main/scala/com/kevel/apso/aws/S3Bucket.scala
+++ b/modules/aws/src/main/scala/com/kevel/apso/aws/S3Bucket.scala
@@ -4,8 +4,9 @@ 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
@@ -13,12 +14,16 @@ 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
+    })
+}
diff --git a/modules/aws/src/test/scala/com/kevel/apso/aws/S3BucketSpec.scala b/modules/aws/src/test/scala/com/kevel/apso/aws/S3BucketSpec.scala
new file mode 100644
index 00000000..4bd2c126
--- /dev/null
+++ b/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
+      }
+    }
+  }
+}
diff --git a/modules/core/src/main/scala/com/kevel/apso/Retry.scala b/modules/core/src/main/scala/com/kevel/apso/Retry.scala
index c630ba9d..43d99554 100644
--- a/modules/core/src/main/scala/com/kevel/apso/Retry.scala
+++ b/modules/core/src/main/scala/com/kevel/apso/Retry.scala
@@ -1,5 +1,8 @@
 package com.kevel.apso
 
+import java.util.concurrent.ThreadLocalRandom
+
+import scala.annotation.tailrec
 import scala.concurrent.duration.*
 import scala.concurrent.{ExecutionContext, Future, blocking}
 import scala.util.control.NonFatal
@@ -73,4 +76,91 @@ object Retry {
     */
   def retry[T](maxRetries: Int = 10, inBetweenSleep: FiniteDuration = 100.millis)(f: => T): Try[T] =
     retry(maxRetries, Option(inBetweenSleep))(f)
+
+  /** Computes the duration to wait before the next attempt, growing exponentially with the number of attempts made.
+    *
+    * @param attempt
+    *   the zero-based index of the attempt that failed
+    * @param base
+    *   the base waiting duration
+    * @param max
+    *   the optional duration with which to cap the exponential growth
+    * @param factor
+    *   the factor of the exponential duration
+    * @param jitter
+    *   the fraction of the 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
+    * @return
+    *   the duration to wait before the next attempt
+    */
+  private[apso] def exponentialBackOffDelay(
+      attempt: Int,
+      base: FiniteDuration,
+      max: Option[FiniteDuration] = None,
+      factor: Double = 2.0,
+      jitter: Double = 1.0
+  ): FiniteDuration = {
+    val exponential = (base.toMillis * Math.pow(factor, attempt.toDouble)).toLong
+    val capped = max.fold(exponential)(m => Math.min(exponential, m.toMillis))
+    val jitterFraction = Math.min(1.0, Math.max(0.0, jitter))
+    (((1.0 - jitterFraction) * capped) + (ThreadLocalRandom
+      .current()
+      .nextDouble() * jitterFraction * capped)).toLong.millis
+  }
+
+  /** Performs a function `f` until it succeeds or until maximum retries is reached, waiting between attempts for a
+    * duration that grows exponentially with the number of attempts made.
+    *
+    * @param maxRetries
+    *   the number of retries
+    * @param base
+    *   the base waiting duration
+    * @param max
+    *   the optional duration with which to cap the exponential growth
+    * @param factor
+    *   the factor of the exponential duration
+    * @param jitter
+    *   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
+    * @param retryWhen
+    *   the predicate deciding whether a failure is worth retrying
+    * @param onRetry
+    *   the function called before each retry with the failure being retried, the duration that will be waited for and
+    *   the number of retries still left
+    * @param onMaxRetriesReached
+    *   the function called when the max retries are reached with the latest failure
+    * @param f
+    *   the function to retry
+    * @return
+    *   a Try of the `f` function result
+    */
+  def exponentialBackOff[T](
+      maxRetries: Int,
+      base: FiniteDuration,
+      max: Option[FiniteDuration] = None,
+      factor: Double = 2.0,
+      jitter: Double = 1.0,
+      retryWhen: Throwable => Boolean = _ => true,
+      onRetry: (Throwable, FiniteDuration, Int) => Unit = (_, _, _) => (),
+      onMaxRetriesReached: Throwable => Unit = _ => ()
+  )(f: => T): Try[T] = {
+    @tailrec
+    def aux(attempt: Int): Try[T] =
+      Try(f) match {
+        case res @ Success(_)      => res
+        case failure @ Failure(ex) =>
+          if (!retryWhen(ex)) failure
+          else if (attempt >= maxRetries) {
+            onMaxRetriesReached(ex)
+            failure
+          } else {
+            val delay = exponentialBackOffDelay(attempt, base, max, factor, jitter)
+            onRetry(ex, delay, maxRetries - attempt)
+            blocking(Thread.sleep(delay.toMillis))
+            aux(attempt + 1)
+          }
+      }
+
+    aux(0)
+  }
 }
diff --git a/modules/core/src/test/scala/com/kevel/apso/RetrySpec.scala b/modules/core/src/test/scala/com/kevel/apso/RetrySpec.scala
index 6fc271bd..63ee2887 100644
--- a/modules/core/src/test/scala/com/kevel/apso/RetrySpec.scala
+++ b/modules/core/src/test/scala/com/kevel/apso/RetrySpec.scala
@@ -1,6 +1,7 @@
 package com.kevel.apso
 
 import scala.concurrent.Future
+import scala.concurrent.duration.*
 import scala.util.Failure
 
 import org.specs2.concurrent.ExecutionEnv
@@ -88,5 +89,221 @@ class RetrySpec(implicit ee: ExecutionEnv) extends Specification {
 
       attempts must beEqualTo(1) // 1 attempt
     }
+
+    "compute an exponentially growing back-off delay" in {
+      val delays =
+        (0 until 4).map(attempt => Retry.exponentialBackOffDelay(attempt, 100.millis, jitter = 0.0))
+
+      delays.map(_.toMillis) must beEqualTo(Seq(100L, 200L, 400L, 800L))
+    }
+
+    "cap the back-off delay with the given maximum" in {
+      val delays =
+        (0 until 4).map(attempt => Retry.exponentialBackOffDelay(attempt, 100.millis, Some(250.millis), jitter = 0.0))
+
+      delays.map(_.toMillis) must beEqualTo(Seq(100L, 200L, 250L, 250L))
+    }
+
+    "grow the back-off delay by the given factor" in {
+      val delays =
+        (0 until 4).map(attempt => Retry.exponentialBackOffDelay(attempt, 100.millis, factor = 3.0, jitter = 0.0))
+
+      delays.map(_.toMillis) must beEqualTo(Seq(100L, 300L, 900L, 2_700L))
+    }
+
+    "randomize the whole back-off delay with a full jitter" in {
+      val delays = (1 to 100).map(_ => Retry.exponentialBackOffDelay(0, 100.millis, jitter = 1.0).toMillis)
+
+      forall(delays)(d => d must beBetween(0L, 100L))
+      delays.distinct.size must be_>(1) // the jitter is random, so the delays must not all be the same
+      delays.min must be_<(50L) // the delays spread over the whole range, instead of clustering near the top
+    }
+
+    "randomize only the given fraction of the back-off delay" in {
+      val delays = (1 to 100).map(_ => Retry.exponentialBackOffDelay(0, 100.millis, jitter = 0.25).toMillis)
+
+      forall(delays)(d => d must beBetween(75L, 100L))
+      delays.distinct.size must be_>(1)
+    }
+
+    "clamp the jitter to a fraction of the back-off delay" in {
+      Retry.exponentialBackOffDelay(0, 100.millis, jitter = -1.0).toMillis must beEqualTo(100L)
+
+      val delays = (1 to 20).map(_ => Retry.exponentialBackOffDelay(0, 100.millis, jitter = 2.0).toMillis)
+      forall(delays)(d => d must beBetween(0L, 100L))
+    }
+
+    "retry a given function with exponential back-off a number of times" in {
+      var attempts = 0
+
+      val f = Retry.exponentialBackOff(10, 1.milli, jitter = 0.0) {
+        attempts = attempts + 1
+        if (attempts <= 3) throw new RuntimeException("Doomed")
+        else attempts
+      }
+
+      f must beSuccessfulTry(4)
+      attempts must beEqualTo(4)
+    }
+
+    "retry a doomed function with exponential back-off until it fails" in {
+      var attempts = 0
+      val retries = 5
+
+      val f = Retry.exponentialBackOff[Any](retries, 1.milli, jitter = 0.0) {
+        attempts = attempts + 1
+        throw new RuntimeException("Doomed")
+      }
+
+      f must beAFailedTry
+      attempts must beEqualTo(1 + retries) // 1 attempt + 5 retries
+    }
+
+    "not retry a failure rejected by the given predicate" in {
+      var attempts = 0
+
+      val f = Retry.exponentialBackOff[Any](
+        10,
+        1.milli,
+        jitter = 0.0,
+        retryWhen = _.getMessage != "Fatal"
+      ) {
+        attempts = attempts + 1
+        throw new RuntimeException("Fatal")
+      }
+
+      f must beAFailedTry
+      attempts must beEqualTo(1) // 1 attempt, no retries
+    }
+
+    "evaluate the given predicate once per failure" in {
+      var seen = 0
+
+      Retry.exponentialBackOff[Any](
+        2,
+        1.milli,
+        jitter = 0.0,
+        retryWhen = { _ =>
+          seen += 1
+          true
+        }
+      ) {
+        throw new RuntimeException("Doomed")
+      }
+
+      seen must beEqualTo(3) // 1 attempt + 2 retries, all failing
+    }
+
+    "report each retry to the given function" in {
+      var retries = List.empty[(String, Long, Int)]
+
+      Retry.exponentialBackOff[Any](
+        3,
+        100.millis,
+        jitter = 0.0,
+        onRetry = (ex, delay, remaining) => retries = retries :+ (ex.getMessage, delay.toMillis, remaining)
+      ) {
+        throw new RuntimeException("Doomed")
+      }
+
+      // one report per retry, never for the final failed attempt
+      retries must beEqualTo(List(("Doomed", 100L, 3), ("Doomed", 200L, 2), ("Doomed", 400L, 1)))
+    }
+
+    "report reaching the maximum retries with the failure that exhausted them" in {
+      var reached = List.empty[String]
+
+      val f = Retry.exponentialBackOff[Any](
+        2,
+        1.milli,
+        jitter = 0.0,
+        onMaxRetriesReached = ex => reached = reached :+ ex.getMessage
+      ) {
+        throw new RuntimeException("Doomed")
+      }
+
+      f must beAFailedTry
+      reached must beEqualTo(List("Doomed")) // reported once, after the last attempt failed
+    }
+
+    "not report reaching the maximum retries when the failure is rejected by the predicate" in {
+      var attempts = 0
+      var retried = 0
+      var reached = 0
+
+      val f = Retry.exponentialBackOff[Any](
+        5,
+        1.milli,
+        jitter = 0.0,
+        retryWhen = _ => false,
+        onRetry = (_, _, _) => retried = retried + 1,
+        onMaxRetriesReached = _ => reached = reached + 1
+      ) {
+        attempts = attempts + 1
+        throw new RuntimeException("Fatal")
+      }
+
+      f must beAFailedTry
+      attempts must beEqualTo(1)
+      retried must beEqualTo(0)
+      reached must beEqualTo(0) // the retries were never exhausted, the failure was rejected upfront
+    }
+
+    "not report reaching the maximum retries when a retry succeeds" in {
+      var attempts = 0
+      var reached = 0
+
+      val f = Retry.exponentialBackOff(
+        5,
+        1.milli,
+        jitter = 0.0,
+        onMaxRetriesReached = _ => reached = reached + 1
+      ) {
+        attempts = attempts + 1
+        if (attempts < 3) throw new RuntimeException("Doomed") else attempts
+      }
+
+      f must beSuccessfulTry(3)
+      reached must beEqualTo(0)
+    }
+
+    "not report anything when the function succeeds on the first attempt" in {
+      var retried = 0
+      var reached = 0
+
+      val f = Retry.exponentialBackOff(
+        5,
+        1.milli,
+        jitter = 0.0,
+        onRetry = (_, _, _) => retried = retried + 1,
+        onMaxRetriesReached = _ => reached = reached + 1
+      )(42)
+
+      f must beSuccessfulTry(42)
+      retried must beEqualTo(0)
+      reached must beEqualTo(0)
+    }
+
+    "make a single attempt when no retries are allowed" in {
+      var attempts = 0
+      var retried = 0
+      var reached = 0
+
+      val f = Retry.exponentialBackOff[Any](
+        0,
+        1.milli,
+        jitter = 0.0,
+        onRetry = (_, _, _) => retried = retried + 1,
+        onMaxRetriesReached = _ => reached = reached + 1
+      ) {
+        attempts = attempts + 1
+        throw new RuntimeException("Doomed")
+      }
+
+      f must beAFailedTry
+      attempts must beEqualTo(1)
+      retried must beEqualTo(0)
+      reached must beEqualTo(1)
+    }
   }
 }
diff --git a/modules/gcp/src/main/scala/com/kevel/apso/gcp/GCSBucket.scala b/modules/gcp/src/main/scala/com/kevel/apso/gcp/GCSBucket.scala
index 301ad81c..c4eb767d 100644
--- a/modules/gcp/src/main/scala/com/kevel/apso/gcp/GCSBucket.scala
+++ b/modules/gcp/src/main/scala/com/kevel/apso/gcp/GCSBucket.scala
@@ -5,14 +5,17 @@ import java.nio.channels.Channels
 import java.nio.file.Path
 
 import scala.annotation.unused
+import scala.concurrent.duration.*
 import scala.jdk.CollectionConverters.*
-import scala.util.{Failure, Success, Try}
+import scala.util.Try
 
 import com.google.cloud.BaseServiceException
 import com.google.cloud.storage.Storage.{BlobListOption, BlobSourceOption, BlobWriteOption}
 import com.google.cloud.storage.{Blob, BlobId, BlobInfo, Storage, StorageException}
+import com.typesafe.config.ConfigFactory
 import com.typesafe.scalalogging.LazyLogging
 
+import com.kevel.apso.Retry
 import com.kevel.apso.gcp.GCSBucket.{noGzipTranscoding, rawInputStream}
 
 final class GCSBucket(
@@ -22,6 +25,18 @@ final class GCSBucket(
     with LazyLogging {
   @transient private[this] lazy val storage: Storage = mkStorage()
 
+  private[this] lazy val config = ConfigFactory.load()
+
+  private[this] lazy val retryPrefix = "gcp.storage.retry"
+  private[this] lazy val retryMaxRetries =
+    Try(config.getInt(retryPrefix + ".max-retries")).getOrElse(GCSBucket.DefaultMaxRetries)
+  private[this] lazy val retryBaseBackOff =
+    Try(config.getDuration(retryPrefix + ".base-backoff").toMillis.millis).getOrElse(GCSBucket.DefaultBaseBackOff)
+  private[this] lazy val retryMaxBackOff =
+    Try(config.getDuration(retryPrefix + ".max-backoff").toMillis.millis).getOrElse(GCSBucket.DefaultMaxBackOff)
+  private[this] lazy val retryJitter =
+    Try(config.getDouble(retryPrefix + ".jitter")).getOrElse(GCSBucket.DefaultJitter)
+
   private def blobId(key: String) = BlobId.of(bucketName, key)
 
   /** Returns size of the file in the location specified by `key` in the bucket. If the file doesn't exist the return
@@ -215,6 +230,9 @@ final class GCSBucket(
     Channels.newInputStream(reader)
   }
 
+  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] = {
     case ex: StorageException =>
       ex.getCode match {
@@ -225,38 +243,43 @@ final class GCSBucket(
           logger.error("No permission to access the file", ex)
           true // no need to retry
         case _ =>
-          logger.warn(
+          log(
+            !ex.isRetryable,
             s"""|GCS service error: ${ex.getMessage}.
                 |Additional details: ${ex.getDebugInfo}""".stripMargin,
             ex
           )
-          false
+          !ex.isRetryable
       }
 
     case ex: BaseServiceException =>
-      logger.warn("An error occurred", ex)
-      ex.isRetryable
+      log(!ex.isRetryable, "An error occurred", ex)
+      !ex.isRetryable
   }
 
-  private[this] def retry[T](f: => T, tries: Int = 3, sleepTime: Int = 5000): Option[T] =
-    if (tries == 0) {
-      logger.error("Max retries reached. Aborting GCS 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 GCS operation. Retrying in ${sleepTime}ms (${tries - 1} more times)")
-            Thread.sleep(sleepTime)
-          }
-          retry(f, tries - 1, sleepTime)
-
-        case _ => None
-      }
+  private[gcp] 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 GCS operation. Retrying in ${delay.toMillis}ms ($remaining more times)"),
+        onMaxRetriesReached = _ => logger.error("Max retries reached. Aborting GCS operation")
+      )(f)
+      .toOption
 }
 
 object GCSBucket {
+  private[gcp] val DefaultMaxRetries = 2
+  private[gcp] val DefaultBaseBackOff = 3.seconds
+  private[gcp] val DefaultMaxBackOff = 30.seconds
+  private[gcp] val DefaultJitter = 1d
+
   // Disable automatic decompression of gzip-encoded objects so that callers receive the raw bytes as stored in GCS.
   // Without this, the GCS client transparently decompresses objects with `Content-Encoding: gzip`, which breaks
   // callers that expect to handle decompression themselves (e.g. when streaming .gz files).