diff --git a/docs/config-description.md b/docs/config-description.md
index be20dbdd6..c41957a2a 100644
--- a/docs/config-description.md
+++ b/docs/config-description.md
@@ -132,10 +132,26 @@ This document provides a list of all available Clio configuration properties in
 ### database.cassandra.request_timeout
 
 - **Required**: False
-- **Type**: int
+- **Type**: double
 - **Default value**: None
-- **Constraints**: The minimum value is `1`. The maximum value is `4294967295`.
-- **Description**: The maximum amount of time in seconds that the system waits for a request to be fetched from the database.
+- **Constraints**: The value must be a positive double number.
+- **Description**: The maximum amount of time in seconds that the system waits for a request to be fetched from the database. Should be set higher than the server side read timeout. If omitted, no request timeout is applied.
+
+### database.cassandra.initial_request_retry_delay
+
+- **Required**: True
+- **Type**: double
+- **Default value**: `0.5`
+- **Constraints**: The value must be a positive double number.
+- **Description**: How long in seconds to wait before the first retry of a database request that failed with a transient error.
+
+### database.cassandra.max_request_retry_delay
+
+- **Required**: True
+- **Type**: double
+- **Default value**: `5`
+- **Constraints**: The value must be a positive double number.
+- **Description**: Upper bound in seconds for the exponential backoff between retries of a database request.
 
 ### database.cassandra.username
 
diff --git a/src/data/BackendInterface.hpp b/src/data/BackendInterface.hpp
index 7d32f1220..6e1e70f69 100644
--- a/src/data/BackendInterface.hpp
+++ b/src/data/BackendInterface.hpp
@@ -4,6 +4,7 @@
 #include "data/LedgerCacheInterface.hpp"
 #include "data/Types.hpp"
 #include "etl/CorruptionDetector.hpp"
+#include "util/Retry.hpp"
 #include "util/Spawn.hpp"
 #include "util/log/Logger.hpp"
 
@@ -35,41 +36,118 @@
 namespace data {
 
 /**
- * @brief Represents a database timeout error.
+ * @brief Represents a transient database error that the caller should retry.
  */
-class DatabaseTimeout : public std::exception {
+class DatabaseError : public std::exception {
+    std::string message_{"Transient database error. Please retry the request"};
+
 public:
+    DatabaseError() = default;
+
+    /**
+     * @brief Construct with a description of the underlying failure.
+     *
+     * @param message What actually went wrong.
+     */
+    explicit DatabaseError(std::string message) : message_{std::move(message)}
+    {
+    }
+
     /**
      * @return The error message as a C string
      */
     [[nodiscard]] char const*
-    what() const throw() override
+    what() const noexcept override
     {
-        return "Database read timed out. Please retry the request";
+        return message_.c_str();
     }
 };
 
-static constexpr std::size_t kDefaultWaitBetweenRetry = 500;
 /**
- * @brief A helper function that catches DatabaseTimeout exceptions and retries indefinitely.
+ * @brief Delay before the first retry in @ref retryOnTimeout().
+ */
+static constexpr std::chrono::milliseconds kDefaultWaitBetweenRetry{500};
+
+/**
+ * @brief Default upper bound for the exponential backoff in @ref retryOnTimeout().
+ */
+static constexpr std::chrono::milliseconds kMaxWaitBetweenRetry{5'000};
+
+/**
+ * @brief Default delays for @ref retryOnTimeout().
+ */
+static constexpr util::Retry::Delays kDefaultRetryDelays{
+    .initial = kDefaultWaitBetweenRetry,
+    .max = kDefaultWaitBetweenRetry
+};
+
+/**
+ * @brief Retry `func` while it throws DatabaseError, suspending the calling coroutine in between.
  *
  * @tparam FnType The type of function object to execute
  * @param func The function object to execute
- * @param waitMs Delay between retry attempts
+ * @param yield The coroutine to suspend between attempts
+ * @param delays The delays to use between attempts
  * @return The same as the return type of func
  */
 template <typename FnType>
 auto
-retryOnTimeout(FnType func, size_t waitMs = kDefaultWaitBetweenRetry)
+retryOnTimeout(
+    FnType func,
+    boost::asio::yield_context yield,
+    util::Retry::Delays delays = kDefaultRetryDelays
+)
 {
     static util::Logger const log{"Backend"};  // NOLINT(readability-identifier-naming)
 
+    auto retry = util::makeRetryExponentialBackoff(delays, yield.get_executor());
+
     while (true) {
         try {
             return func();
-        } catch (DatabaseTimeout const&) {
-            LOG(log.error()) << "Database request timed out. Sleeping and retrying ... ";
-            std::this_thread::sleep_for(std::chrono::milliseconds(waitMs));
+        } catch (DatabaseError const& e) {
+            auto const delayMs =
+                std::chrono::duration_cast<std::chrono::milliseconds>(retry.delayValue()).count();
+            LOG(log.error()) << e.what() << " (attempt " << retry.attemptNumber() + 1
+                             << "). Retrying in " << delayMs << "ms ...";
+
+            retry.wait(yield);
+        }
+    }
+}
+
+/**
+ * @brief Retry `func` while it throws DatabaseError, blocking the calling thread in between.
+ *
+ * @warning Blocks the calling thread; from a coroutine use the `yield_context` overload instead.
+ *
+ * @tparam FnType The type of function object to execute
+ * @param func The function object to execute
+ * @param delays The delays to use between attempts
+ * @return The same as the return type of func
+ */
+template <typename FnType>
+auto
+retryOnTimeout(FnType func, util::Retry::Delays delays = kDefaultRetryDelays)
+{
+    static util::Logger const log{"Backend"};  // NOLINT(readability-identifier-naming)
+
+    util::ExponentialBackoffStrategy backoff{delays};
+    std::size_t attempt = 1;
+
+    while (true) {
+        try {
+            return func();
+        } catch (DatabaseError const& e) {
+            auto const delay = backoff.getDelay();
+            LOG(log.error()) << e.what() << " (attempt " << attempt << "). Retrying in "
+                             << std::chrono::duration_cast<std::chrono::milliseconds>(delay).count()
+                             << "ms ...";
+
+            ++attempt;
+
+            std::this_thread::sleep_for(delay);
+            backoff.increaseDelay();
         }
     }
 }
@@ -105,18 +183,21 @@ synchronous(FnType&& func)
 }
 
 /**
- * @brief Synchronously execute the given function object and retry until no DatabaseTimeout is
+ * @brief Synchronously execute the given function object and retry until no DatabaseError is
  * thrown.
  *
+ * @warning Blocks the calling thread while backing off.
+ *
  * @tparam FnType The type of function object to execute
  * @param func The function object to execute
+ * @param delays The delays to use between attempts
  * @return The same as the return type of func
  */
 template <typename FnType>
 auto
-synchronousAndRetryOnTimeout(FnType&& func)
+synchronousAndRetryOnTimeout(FnType&& func, util::Retry::Delays delays = kDefaultRetryDelays)
 {
-    return retryOnTimeout([&]() { return synchronous(func); });
+    return retryOnTimeout([&]() { return synchronous(func); }, delays);
 }
 
 /**
@@ -139,8 +220,27 @@ public:
     BackendInterface(LedgerCacheInterface& cache) : cache_{cache}
     {
     }
+
     virtual ~BackendInterface() = default;
 
+    /**
+     * @return Delay before the first retry of a request against this backend
+     */
+    [[nodiscard]] virtual std::chrono::milliseconds
+    initialRetryDelay() const
+    {
+        return kDefaultWaitBetweenRetry;
+    }
+
+    /**
+     * @return Upper bound for the retry backoff; equal to @ref initialRetryDelay() means flat
+     */
+    [[nodiscard]] virtual std::chrono::milliseconds
+    maxRetryDelay() const
+    {
+        return kMaxWaitBetweenRetry;
+    }
+
     // TODO https://github.com/XRPLF/clio/issues/1956: Remove this hack once old ETL is removed.
     // Cache should not be exposed thru BackendInterface
 
@@ -705,7 +805,7 @@ public:
     hardFetchLedgerRange(boost::asio::yield_context yield) const = 0;
 
     /**
-     * @brief Fetches the ledger range from DB retrying until no DatabaseTimeout is thrown.
+     * @brief Fetches the ledger range from DB retrying until no DatabaseError is thrown.
      *
      * @return The ledger range if available; nullopt otherwise
      */
diff --git a/src/data/cassandra/CassandraBackendFamily.hpp b/src/data/cassandra/CassandraBackendFamily.hpp
index 3d0aa01f3..953daedff 100644
--- a/src/data/cassandra/CassandraBackendFamily.hpp
+++ b/src/data/cassandra/CassandraBackendFamily.hpp
@@ -139,6 +139,24 @@ public:
      */
     CassandraBackendFamily(CassandraBackendFamily&&) = delete;
 
+    /**
+     * @return The configured delay before the first retry
+     */
+    [[nodiscard]] std::chrono::milliseconds
+    initialRetryDelay() const override
+    {
+        return settingsProvider_.getInitialRetryDelay();
+    }
+
+    /**
+     * @return The configured upper bound for the retry backoff
+     */
+    [[nodiscard]] std::chrono::milliseconds
+    maxRetryDelay() const override
+    {
+        return settingsProvider_.getMaxRetryDelay();
+    }
+
     TransactionsAndCursor
     fetchAccountTransactions(
         xrpl::AccountID const& account,
diff --git a/src/data/cassandra/Error.hpp b/src/data/cassandra/Error.hpp
index 88709ef47..58033ae20 100644
--- a/src/data/cassandra/Error.hpp
+++ b/src/data/cassandra/Error.hpp
@@ -80,17 +80,6 @@ public:
         return code_;
     }
 
-    /**
-     * @return true if the wrapped error is considered a timeout; false otherwise
-     */
-    [[nodiscard]] bool
-    isTimeout() const
-    {
-        return code_ == CASS_ERROR_LIB_NO_HOSTS_AVAILABLE or
-            code_ == CASS_ERROR_LIB_REQUEST_TIMED_OUT or code_ == CASS_ERROR_SERVER_UNAVAILABLE or
-            code_ == CASS_ERROR_SERVER_OVERLOADED or code_ == CASS_ERROR_SERVER_READ_TIMEOUT;
-    }
-
     /**
      * @return true if the wrapped error is an invalid query; false otherwise
      */
diff --git a/src/data/cassandra/SettingsProvider.cpp b/src/data/cassandra/SettingsProvider.cpp
index d0f57eaa3..1284c8e24 100644
--- a/src/data/cassandra/SettingsProvider.cpp
+++ b/src/data/cassandra/SettingsProvider.cpp
@@ -3,6 +3,7 @@
 #include "data/cassandra/Types.hpp"
 #include "data/cassandra/impl/Cluster.hpp"
 #include "util/Constants.hpp"
+#include "util/config/ConfigDefinition.hpp"
 #include "util/config/ObjectView.hpp"
 
 #include <cerrno>
@@ -25,6 +26,12 @@ SettingsProvider::SettingsProvider(util::config::ObjectView const& cfg)
     , keyspace_{cfg.get<std::string>("keyspace")}
     , tablePrefix_{cfg.maybeValue<std::string>("table_prefix")}
     , replicationFactor_{cfg.get<uint16_t>("replication_factor")}
+    , initialRetryDelay_{util::config::ClioConfigDefinition::toMilliseconds(
+          cfg.get<float>("initial_request_retry_delay")
+      )}
+    , maxRetryDelay_{util::config::ClioConfigDefinition::toMilliseconds(
+          cfg.get<float>("max_request_retry_delay")
+      )}
     , settings_{parseSettings()}
 {
 }
@@ -94,9 +101,9 @@ SettingsProvider::parseSettings() const
     }
 
     if (config_.getValueView("request_timeout").hasValue()) {
-        auto const requestTimeoutSecond = config_.get<uint32_t>("request_timeout");
-        settings.requestTimeout =
-            std::chrono::milliseconds{requestTimeoutSecond * util::kMillisecondsPerSecond};
+        settings.requestTimeout = util::config::ClioConfigDefinition::toMilliseconds(
+            config_.get<float>("request_timeout")
+        );
     }
 
     settings.certificate = parseOptionalCertificate();
diff --git a/src/data/cassandra/SettingsProvider.hpp b/src/data/cassandra/SettingsProvider.hpp
index 631f0401e..34625280e 100644
--- a/src/data/cassandra/SettingsProvider.hpp
+++ b/src/data/cassandra/SettingsProvider.hpp
@@ -4,6 +4,7 @@
 #include "data/cassandra/impl/Cluster.hpp"
 #include "util/config/ObjectView.hpp"
 
+#include <chrono>
 #include <cstdint>
 #include <optional>
 #include <string>
@@ -19,6 +20,8 @@ class SettingsProvider {
     std::string keyspace_;
     std::optional<std::string> tablePrefix_;
     uint16_t replicationFactor_;
+    std::chrono::milliseconds initialRetryDelay_;
+    std::chrono::milliseconds maxRetryDelay_;
     Settings settings_;
 
 public:
@@ -62,6 +65,24 @@ public:
         return replicationFactor_;
     }
 
+    /**
+     * @return Delay before the first retry of a failed request
+     */
+    [[nodiscard]] std::chrono::milliseconds
+    getInitialRetryDelay() const
+    {
+        return initialRetryDelay_;
+    }
+
+    /**
+     * @return Upper bound for the retry backoff
+     */
+    [[nodiscard]] std::chrono::milliseconds
+    getMaxRetryDelay() const
+    {
+        return maxRetryDelay_;
+    }
+
 private:
     [[nodiscard]] std::optional<std::string>
     parseOptionalCertificate() const;
diff --git a/src/data/cassandra/impl/ExecutionStrategy.hpp b/src/data/cassandra/impl/ExecutionStrategy.hpp
index e8907a5f9..6496fcb5e 100644
--- a/src/data/cassandra/impl/ExecutionStrategy.hpp
+++ b/src/data/cassandra/impl/ExecutionStrategy.hpp
@@ -15,6 +15,7 @@
 #include <boost/asio/io_context.hpp>
 #include <boost/asio/spawn.hpp>
 #include <boost/json/object.hpp>
+#include <fmt/format.h>
 
 #include <algorithm>
 #include <atomic>
@@ -27,6 +28,7 @@
 #include <mutex>
 #include <optional>
 #include <stdexcept>
+#include <string>
 #include <thread>
 #include <type_traits>
 #include <vector>
@@ -169,7 +171,7 @@ public:
      *
      * @param preparedStatement Statement to prepare and execute
      * @param args Args to bind to the prepared statement
-     * @throws DatabaseTimeout on timeout
+     * @throws DatabaseError on a database error
      */
     template <typename... Args>
     void
@@ -185,7 +187,7 @@ public:
      * Retries forever with retry policy specified by @ref AsyncExecutor
      *
      * @param statement Statement to execute
-     * @throws DatabaseTimeout on timeout
+     * @throws DatabaseError on a database error
      */
     void
     write(StatementType&& statement)
@@ -215,7 +217,7 @@ public:
      * Retries forever with retry policy specified by @ref AsyncExecutor.
      *
      * @param statements Vector of statements to execute as a batch
-     * @throws DatabaseTimeout on timeout
+     * @throws DatabaseError on a database error
      */
     void
     write(std::vector<StatementType>&& statements)
@@ -254,7 +256,7 @@ public:
      * Retries forever with retry policy specified by @ref AsyncExecutor.
      *
      * @param statements Vector of statements to execute
-     * @throws DatabaseTimeout on timeout
+     * @throws DatabaseError on a database error
      */
     void
     writeEach(std::vector<StatementType>&& statements)
@@ -272,7 +274,7 @@ public:
      * @param token Completion token (yield_context)
      * @param preparedStatement Statement to prepare and execute
      * @param args Args to bind to the prepared statement
-     * @throws DatabaseTimeout on timeout
+     * @throws DatabaseError on a database error
      * @return ResultType or error wrapped in Expected
      */
     template <typename... Args>
@@ -289,7 +291,7 @@ public:
      *
      * @param token Completion token (yield_context)
      * @param statements Statements to execute in a batch
-     * @throws DatabaseTimeout on timeout
+     * @throws DatabaseError on a database error
      * @return ResultType or error wrapped in Expected
      */
     [[maybe_unused]] ResultOrErrorType
@@ -346,7 +348,7 @@ public:
      *
      * @param token Completion token (yield_context)
      * @param statement Statement to execute
-     * @throws DatabaseTimeout on timeout
+     * @throws DatabaseError on a database error
      * @return ResultType or error wrapped in Expected
      */
     [[maybe_unused]] ResultOrErrorType
@@ -402,7 +404,7 @@ public:
      *
      * @param token Completion token (yield_context)
      * @param statements Statements to execute
-     * @throws DatabaseTimeout on db error
+     * @throws DatabaseError on a database error
      * @return Vector of results
      */
     std::vector<ResultType>
@@ -457,7 +459,7 @@ public:
             );
             counters_->registerReadError(errorsCount);
             counters_->registerReadFinished(startTime, statements.size() - errorsCount);
-            throw DatabaseTimeout{};
+            throw DatabaseError{};
         }
         counters_->registerReadFinished(startTime, statements.size());
 
@@ -551,11 +553,13 @@ private:
     void
     throwErrorIfNeeded(CassandraError err) const
     {
-        if (err.isTimeout())
-            throw DatabaseTimeout();
-
+        // NOTE: etl::impl::Loader and etl::impl::Extractor treat std::runtime_error as
+        // "amendment blocked", so only genuinely permanent failures may be thrown as one.
         if (err.isInvalidQuery())
             throw std::runtime_error("Invalid query");
+
+        // anything else, including unclassified codes, is transient and gets retried
+        throw DatabaseError{fmt::format("Database error [{}]: {}", err.code(), err.message())};
     }
 };
 
diff --git a/src/data/cassandra/impl/RetryPolicy.hpp b/src/data/cassandra/impl/RetryPolicy.hpp
index 50b664772..6d31581da 100644
--- a/src/data/cassandra/impl/RetryPolicy.hpp
+++ b/src/data/cassandra/impl/RetryPolicy.hpp
@@ -28,8 +28,7 @@ public:
     ExponentialBackoffRetryPolicy(boost::asio::io_context& ioc)
         : retry_(
               util::makeRetryExponentialBackoff(
-                  std::chrono::milliseconds(1),
-                  std::chrono::seconds(1),
+                  {.initial = std::chrono::milliseconds(1), .max = std::chrono::seconds(1)},
                   boost::asio::make_strand(ioc)
               )
           )
diff --git a/src/etl/impl/CacheLoader.hpp b/src/etl/impl/CacheLoader.hpp
index de36ac570..1db69775d 100644
--- a/src/etl/impl/CacheLoader.hpp
+++ b/src/etl/impl/CacheLoader.hpp
@@ -145,9 +145,17 @@ private:
             LOG(log_.debug()) << "Starting a cursor: " << xrpl::strHex(start);
 
             while (not token.isStopRequested() and not cache_.get().isDisabled()) {
-                auto res = data::retryOnTimeout([this, seq, cachePageFetchSize, &start, token]() {
-                    return backend_->fetchLedgerPage(start, seq, cachePageFetchSize, false, token);
-                });
+                auto res = data::retryOnTimeout(
+                    [this, seq, cachePageFetchSize, &start, token]() {
+                        return backend_->fetchLedgerPage(
+                            start, seq, cachePageFetchSize, false, token
+                        );
+                    },
+                    token,
+                    util::Retry::Delays{
+                        .initial = backend_->initialRetryDelay(), .max = backend_->maxRetryDelay()
+                    }
+                );
 
                 cache_.get().update(res.objects, seq, true);
 
diff --git a/src/etl/impl/SubscriptionSource.cpp b/src/etl/impl/SubscriptionSource.cpp
index 833d3ac04..56fc3be6b 100644
--- a/src/etl/impl/SubscriptionSource.cpp
+++ b/src/etl/impl/SubscriptionSource.cpp
@@ -58,7 +58,9 @@ SubscriptionSource::SubscriptionSource(
     , subscriptions_(std::move(subscriptions))
     , strand_(boost::asio::make_strand(ioContext))
     , wsTimeout_(wsTimeout)
-    , retry_(util::makeRetryExponentialBackoff(retryDelay, kRetryMaxDelay, strand_))
+    , retry_(
+          util::makeRetryExponentialBackoff({.initial = retryDelay, .max = kMaxRetryDelay}, strand_)
+      )
     , onConnect_(std::move(onConnect))
     , onDisconnect_(std::move(onDisconnect))
     , onLedgerClosed_(std::move(onLedgerClosed))
diff --git a/src/etl/impl/SubscriptionSource.hpp b/src/etl/impl/SubscriptionSource.hpp
index 1e35181fe..08cea70f0 100644
--- a/src/etl/impl/SubscriptionSource.hpp
+++ b/src/etl/impl/SubscriptionSource.hpp
@@ -75,7 +75,7 @@ private:
     util::StopHelper stopHelper_;
 
     static constexpr std::chrono::seconds kWsTimeout{30};
-    static constexpr std::chrono::seconds kRetryMaxDelay{30};
+    static constexpr std::chrono::seconds kMaxRetryDelay{30};
     static constexpr std::chrono::seconds kRetryDelay{1};
 
 public:
diff --git a/src/rpc/Errors.hpp b/src/rpc/Errors.hpp
index 8f69d8157..5b06c365a 100644
--- a/src/rpc/Errors.hpp
+++ b/src/rpc/Errors.hpp
@@ -239,7 +239,7 @@ public:
      * @return The error message
      */
     [[nodiscard]] char const*
-    what() const throw() override
+    what() const noexcept override
     {
         return msg_.c_str();
     }
@@ -267,7 +267,7 @@ public:
      * @return The error message
      */
     [[nodiscard]] char const*
-    what() const throw() override
+    what() const noexcept override
     {
         return account_.c_str();
     }
diff --git a/src/rpc/RPCEngine.hpp b/src/rpc/RPCEngine.hpp
index c6380ba9f..a44f33707 100644
--- a/src/rpc/RPCEngine.hpp
+++ b/src/rpc/RPCEngine.hpp
@@ -178,8 +178,8 @@ public:
             }
 
             return Result{std::move(v)};
-        } catch (data::DatabaseTimeout const& t) {
-            LOG(log_.error()) << "Database timeout";
+        } catch (data::DatabaseError const& t) {
+            LOG(log_.error()) << "Database error: " << t.what();
             notifyTooBusy();
 
             return Result{Status{RippledError::RpcTooBusy}};
@@ -361,8 +361,8 @@ private:
             }
 
             return Result{std::move(v)};
-        } catch (data::DatabaseTimeout const& t) {
-            LOG(log_.error()) << "Database timeout";
+        } catch (data::DatabaseError const& t) {
+            LOG(log_.error()) << "Database error: " << t.what();
             notifyTooBusy();
 
             return Result{Status{RippledError::RpcTooBusy}};
diff --git a/src/util/Retry.cpp b/src/util/Retry.cpp
index 29d2fdba8..d45a1f2d2 100644
--- a/src/util/Retry.cpp
+++ b/src/util/Retry.cpp
@@ -1,6 +1,8 @@
 #include "util/Retry.hpp"
 
+#include <boost/asio/any_io_executor.hpp>
 #include <boost/asio/io_context.hpp>
+#include <boost/asio/spawn.hpp>
 #include <boost/asio/strand.hpp>
 
 #include <algorithm>
@@ -42,6 +44,11 @@ Retry::Retry(
 {
 }
 
+Retry::Retry(RetryStrategyPtr strategy, boost::asio::any_io_executor executor)
+    : strategy_(std::move(strategy)), timer_(executor)
+{
+}
+
 Retry::~Retry()
 {
     *canceled_ = true;
@@ -73,11 +80,8 @@ Retry::reset()
     (*strategy_).reset();
 }
 
-ExponentialBackoffStrategy::ExponentialBackoffStrategy(
-    std::chrono::steady_clock::duration delay,
-    std::chrono::steady_clock::duration maxDelay
-)
-    : RetryStrategy(delay), maxDelay_(maxDelay)
+ExponentialBackoffStrategy::ExponentialBackoffStrategy(Retry::Delays delays)
+    : RetryStrategy(delays.initial), maxDelay_(delays.max)
 {
 }
 
@@ -88,14 +92,32 @@ ExponentialBackoffStrategy::nextDelay() const
     return std::min(next, maxDelay_);
 }
 
+void
+Retry::wait(boost::asio::yield_context yield)
+{
+    *canceled_ = false;
+    timer_.expires_after(strategy_->getDelay());
+    strategy_->increaseDelay();
+    ++attemptNumber_;
+
+    // error ignored on purpose: a cancelled timer just means the caller retries sooner
+    boost::system::error_code ec;
+    timer_.async_wait(yield[ec]);
+}
+
 Retry
 makeRetryExponentialBackoff(
-    std::chrono::steady_clock::duration delay,
-    std::chrono::steady_clock::duration maxDelay,
+    Retry::Delays delays,
     boost::asio::strand<boost::asio::io_context::executor_type> strand
 )
 {
-    return Retry(std::make_unique<ExponentialBackoffStrategy>(delay, maxDelay), std::move(strand));
+    return Retry(std::make_unique<ExponentialBackoffStrategy>(delays), std::move(strand));
+}
+
+Retry
+makeRetryExponentialBackoff(Retry::Delays delays, boost::asio::any_io_executor executor)
+{
+    return Retry(std::make_unique<ExponentialBackoffStrategy>(delays), std::move(executor));
 }
 
 }  // namespace util
diff --git a/src/util/Retry.hpp b/src/util/Retry.hpp
index 3064423e2..b58ec06b2 100644
--- a/src/util/Retry.hpp
+++ b/src/util/Retry.hpp
@@ -1,7 +1,9 @@
 #pragma once
 
+#include <boost/asio/any_io_executor.hpp>
 #include <boost/asio/error.hpp>
 #include <boost/asio/io_context.hpp>
+#include <boost/asio/spawn.hpp>
 #include <boost/asio/steady_timer.hpp>
 #include <boost/asio/strand.hpp>
 
@@ -65,6 +67,16 @@ class Retry {
     std::shared_ptr<std::atomic_bool> canceled_{std::make_shared<std::atomic_bool>(false)};
 
 public:
+    /**
+     * @brief The delays to use between retry attempts
+     *
+     * Equal `initial` and `max` mean a flat delay with no backoff.
+     */
+    struct Delays {
+        std::chrono::steady_clock::duration initial;
+        std::chrono::steady_clock::duration max;
+    };
+
     /**
      * @brief Construct a new Retry object
      *
@@ -76,6 +88,16 @@ public:
         boost::asio::strand<boost::asio::io_context::executor_type> strand
     );
 
+    /**
+     * @brief Construct a new Retry object from any I/O executor
+     *
+     * For coroutines, pass `yield.get_executor()` and drive it with @ref wait().
+     *
+     * @param strategy The retry strategy to use
+     * @param executor The executor to run the retry timer on
+     */
+    Retry(RetryStrategyPtr strategy, boost::asio::any_io_executor executor);
+
     /**
      * @brief Destroy the Retry object
      */
@@ -105,6 +127,17 @@ public:
         });
     }
 
+    /**
+     * @brief Wait out the current delay by suspending the calling coroutine, then back off.
+     *
+     * Unlike @ref retry() this returns once the delay elapsed instead of scheduling a callback, so
+     * the caller can keep its own loop. Advances the delay and attempt number like @ref retry().
+     *
+     * @param yield The coroutine to suspend
+     */
+    void
+    wait(boost::asio::yield_context yield);
+
     /**
      * @brief Cancel scheduled retry if any
      */
@@ -140,13 +173,9 @@ public:
     /**
      * @brief Construct a new Exponential Backoff Strategy object
      *
-     * @param delay The initial delay value
-     * @param maxDelay The maximum delay value
+     * @param delays The delays to use between attempts
      */
-    ExponentialBackoffStrategy(
-        std::chrono::steady_clock::duration delay,
-        std::chrono::steady_clock::duration maxDelay
-    );
+    explicit ExponentialBackoffStrategy(Retry::Delays delays);
 
 private:
     [[nodiscard]] std::chrono::steady_clock::duration
@@ -156,16 +185,24 @@ private:
 /**
  * @brief Create a retry mechanism with exponential backoff strategy
  *
- * @param delay The initial delay value
- * @param maxDelay The maximum delay value
+ * @param delays The delays to use between attempts
  * @param strand The strand to use for async operations
  * @return The retry object
  */
 Retry
 makeRetryExponentialBackoff(
-    std::chrono::steady_clock::duration delay,
-    std::chrono::steady_clock::duration maxDelay,
+    Retry::Delays delays,
     boost::asio::strand<boost::asio::io_context::executor_type> strand
 );
 
+/**
+ * @brief Create a retry mechanism with exponential backoff strategy on any I/O executor
+ *
+ * @param delays The delays to use between attempts
+ * @param executor The executor to run the retry timer on
+ * @return The retry object
+ */
+Retry
+makeRetryExponentialBackoff(Retry::Delays delays, boost::asio::any_io_executor executor);
+
 }  // namespace util
diff --git a/src/util/config/ConfigDefinition.cpp b/src/util/config/ConfigDefinition.cpp
index 7166067fe..0fef3d0a0 100644
--- a/src/util/config/ConfigDefinition.cpp
+++ b/src/util/config/ConfigDefinition.cpp
@@ -278,7 +278,15 @@ getClioConfig()
          {"database.cassandra.connect_timeout",
           ConfigValue{ConfigType::Integer}.optional().withConstraint(gValidateUint32)},
          {"database.cassandra.request_timeout",
-          ConfigValue{ConfigType::Integer}.optional().withConstraint(gValidateUint32)},
+          ConfigValue{ConfigType::Double}.optional().withConstraint(gValidatePositiveDouble)},
+         {"database.cassandra.initial_request_retry_delay",
+          ConfigValue{ConfigType::Double}.defaultValue(0.5).withConstraint(
+              gValidatePositiveDouble
+          )},
+         {"database.cassandra.max_request_retry_delay",
+          ConfigValue{ConfigType::Double}.defaultValue(5.0).withConstraint(
+              gValidatePositiveDouble
+          )},
          {"database.cassandra.username", ConfigValue{ConfigType::String}.optional()},
          {"database.cassandra.password", ConfigValue{ConfigType::String}.optional()},
          {"database.cassandra.certfile", ConfigValue{ConfigType::String}.optional()},
diff --git a/src/util/config/ConfigDescription.hpp b/src/util/config/ConfigDescription.hpp
index b0fee6c03..eb6fcd666 100644
--- a/src/util/config/ConfigDescription.hpp
+++ b/src/util/config/ConfigDescription.hpp
@@ -182,8 +182,14 @@ This document provides a list of all available Clio configuration properties in
                     "established."},
         KV{.key = "database.cassandra.request_timeout",
            .value = "The maximum amount of time in seconds that the system waits for a request to "
-                    "be fetched from the "
-                    "database."},
+                    "be fetched from the database. Should be set higher than the server side read "
+                    "timeout. If omitted, no request timeout is applied."},
+        KV{.key = "database.cassandra.initial_request_retry_delay",
+           .value = "How long in seconds to wait before the first retry of a database request "
+                    "that failed with a transient error."},
+        KV{.key = "database.cassandra.max_request_retry_delay",
+           .value = "Upper bound in seconds for the exponential backoff between retries of a "
+                    "database request."},
         KV{.key = "database.cassandra.username",
            .value = "The username used for authenticating with the database."},
         KV{.key = "database.cassandra.password",
diff --git a/tests/integration/data/BackendFactoryTests.cpp b/tests/integration/data/BackendFactoryTests.cpp
index 6dd6cc875..cadb5f82b 100644
--- a/tests/integration/data/BackendFactoryTests.cpp
+++ b/tests/integration/data/BackendFactoryTests.cpp
@@ -52,7 +52,11 @@ protected:
         {"database.cassandra.write_batch_size", ConfigValue{ConfigType::Integer}.defaultValue(20)},
         {"database.cassandra.connect_timeout",
          ConfigValue{ConfigType::Integer}.defaultValue(1).optional()},
-        {"database.cassandra.request_timeout", ConfigValue{ConfigType::Integer}.optional()},
+        {"database.cassandra.initial_request_retry_delay",
+         ConfigValue{ConfigType::Double}.defaultValue(0.5)},
+        {"database.cassandra.max_request_retry_delay",
+         ConfigValue{ConfigType::Double}.defaultValue(5.0)},
+        {"database.cassandra.request_timeout", ConfigValue{ConfigType::Double}.optional()},
         {"database.cassandra.username", ConfigValue{ConfigType::String}.optional()},
         {"database.cassandra.password", ConfigValue{ConfigType::String}.optional()},
         {"database.cassandra.certfile", ConfigValue{ConfigType::String}.optional()},
diff --git a/tests/integration/data/cassandra/BackendTests.cpp b/tests/integration/data/cassandra/BackendTests.cpp
index 3a925e714..d31b160da 100644
--- a/tests/integration/data/cassandra/BackendTests.cpp
+++ b/tests/integration/data/cassandra/BackendTests.cpp
@@ -96,8 +96,12 @@ protected:
         {"database.cassandra.write_batch_size", ConfigValue{ConfigType::Integer}.defaultValue(20)},
         {"database.cassandra.connect_timeout",
          ConfigValue{ConfigType::Integer}.defaultValue(10).optional()},
+        {"database.cassandra.initial_request_retry_delay",
+         ConfigValue{ConfigType::Double}.defaultValue(0.5)},
+        {"database.cassandra.max_request_retry_delay",
+         ConfigValue{ConfigType::Double}.defaultValue(5.0)},
         {"database.cassandra.request_timeout",
-         ConfigValue{ConfigType::Integer}.defaultValue(10).optional()},
+         ConfigValue{ConfigType::Double}.defaultValue(10.0).optional()},
         {"database.cassandra.username", ConfigValue{ConfigType::String}.optional()},
         {"database.cassandra.password", ConfigValue{ConfigType::String}.optional()},
         {"database.cassandra.certfile", ConfigValue{ConfigType::String}.optional()},
diff --git a/tests/integration/migration/cassandra/CassandraMigrationManagerTests.cpp b/tests/integration/migration/cassandra/CassandraMigrationManagerTests.cpp
index 9aa8ad5f3..fbb25a8d4 100644
--- a/tests/integration/migration/cassandra/CassandraMigrationManagerTests.cpp
+++ b/tests/integration/migration/cassandra/CassandraMigrationManagerTests.cpp
@@ -117,8 +117,12 @@ protected:
           ConfigValue{ConfigType::Integer}.defaultValue(20).withConstraint(gValidateUint16)},
          {"database.cassandra.connect_timeout",
           ConfigValue{ConfigType::Integer}.optional().withConstraint(gValidateUint32)},
+         {"database.cassandra.initial_request_retry_delay",
+          ConfigValue{ConfigType::Double}.defaultValue(0.5)},
+         {"database.cassandra.max_request_retry_delay",
+          ConfigValue{ConfigType::Double}.defaultValue(5.0)},
          {"database.cassandra.request_timeout",
-          ConfigValue{ConfigType::Integer}.optional().withConstraint(gValidateUint32)},
+          ConfigValue{ConfigType::Double}.optional().withConstraint(gValidatePositiveDouble)},
          {"database.cassandra.username", ConfigValue{ConfigType::String}.optional()},
          {"database.cassandra.password", ConfigValue{ConfigType::String}.optional()},
          {"database.cassandra.certfile", ConfigValue{ConfigType::String}.optional()},
diff --git a/tests/unit/CMakeLists.txt b/tests/unit/CMakeLists.txt
index dc805b193..e67d37663 100644
--- a/tests/unit/CMakeLists.txt
+++ b/tests/unit/CMakeLists.txt
@@ -15,6 +15,7 @@ target_sources(
         data/LedgerCacheLoadingStateTests.cpp
         data/LedgerCacheSaverTests.cpp
         data/cassandra/AsyncExecutorTests.cpp
+        data/cassandra/ErrorTests.cpp
         data/cassandra/ExecutionStrategyTests.cpp
         data/cassandra/LedgerHeaderCacheTests.cpp
         data/cassandra/RetryPolicyTests.cpp
diff --git a/tests/unit/data/BackendInterfaceTests.cpp b/tests/unit/data/BackendInterfaceTests.cpp
index 19e3267a6..c69dd7b18 100644
--- a/tests/unit/data/BackendInterfaceTests.cpp
+++ b/tests/unit/data/BackendInterfaceTests.cpp
@@ -1,10 +1,13 @@
+#include "data/BackendInterface.hpp"
 #include "etl/CorruptionDetector.hpp"
 #include "etl/SystemState.hpp"
 #include "util/AsioContextTestFixture.hpp"
 #include "util/MockBackendTestFixture.hpp"
 #include "util/MockPrometheus.hpp"
+#include "util/Retry.hpp"
 #include "util/TestObject.hpp"
 
+#include <boost/asio/post.hpp>
 #include <gmock/gmock.h>
 #include <gtest/gtest.h>
 #include <xrpl/basics/Blob.h>
@@ -12,7 +15,12 @@
 #include <xrpl/protocol/Indexes.h>
 #include <xrpl/protocol/XRPAmount.h>
 
+#include <chrono>
+#include <cstddef>
+#include <exception>
 #include <optional>
+#include <stdexcept>
+#include <type_traits>
 #include <vector>
 
 using namespace data;
@@ -160,3 +168,122 @@ TEST_F(
     });
     EXPECT_FALSE(backend_->cache().isDisabled());
 }
+
+// Loader and Extractor catch std::runtime_error to decide the server must amendment-block, so
+// DatabaseError has to stay outside that hierarchy.
+TEST(BackendInterfaceRetryTest, DatabaseErrorIsNotARuntimeError)
+{
+    static_assert(std::is_base_of_v<std::exception, DatabaseError>);
+    static_assert(not std::is_base_of_v<std::runtime_error, DatabaseError>);
+
+    try {
+        throw DatabaseError{"transient"};
+    } catch (std::runtime_error const&) {
+        FAIL() << "DatabaseError must not be caught as std::runtime_error - doing so would "
+                  "amendment-block the server on a transient database error";
+    } catch (std::exception const& e) {
+        EXPECT_STREQ(e.what(), "transient");
+    }
+}
+
+TEST(BackendInterfaceRetryTest, DatabaseErrorKeepsDefaultMessage)
+{
+    EXPECT_STREQ(DatabaseError{}.what(), "Transient database error. Please retry the request");
+}
+
+TEST(BackendInterfaceRetryTest, RetryOnTimeoutBlockingRetriesUntilSuccess)
+{
+    std::size_t calls = 0;
+
+    auto const result = retryOnTimeout(
+        [&calls]() -> int {
+            if (++calls < 3)
+                throw DatabaseError{};
+            return 42;
+        },
+        util::Retry::Delays{
+            .initial = std::chrono::milliseconds{1}, .max = std::chrono::milliseconds{2}
+        }
+    );
+
+    EXPECT_EQ(result, 42);
+    EXPECT_EQ(calls, 3);
+}
+
+TEST(BackendInterfaceRetryTest, RetryOnTimeoutBlockingDoesNotSwallowOtherExceptions)
+{
+    EXPECT_THROW(
+        retryOnTimeout(
+            []() -> int { throw std::runtime_error{"permanent"}; },
+            util::Retry::Delays{
+                .initial = std::chrono::milliseconds{1}, .max = std::chrono::milliseconds{1}
+            }
+        ),
+        std::runtime_error
+    );
+}
+
+struct BackendInterfaceRetryCoroTest : SyncAsioContextTest {};
+
+TEST_F(BackendInterfaceRetryCoroTest, RetryOnTimeoutCoroRetriesUntilSuccess)
+{
+    std::size_t calls = 0;
+
+    runSpawn([&calls](auto yield) {
+        auto const result = retryOnTimeout(
+            [&calls]() -> int {
+                if (++calls < 3)
+                    throw DatabaseError{};
+                return 42;
+            },
+            yield,
+            util::Retry::Delays{
+                .initial = std::chrono::milliseconds{1}, .max = std::chrono::milliseconds{2}
+            }
+        );
+
+        EXPECT_EQ(result, 42);
+    });
+
+    EXPECT_EQ(calls, 3);
+}
+
+TEST_F(BackendInterfaceRetryCoroTest, RetryOnTimeoutCoroDoesNotSwallowOtherExceptions)
+{
+    runSpawn([](auto yield) {
+        EXPECT_THROW(
+            retryOnTimeout(
+                []() -> int { throw std::runtime_error{"permanent"}; },
+                yield,
+                util::Retry::Delays{
+                    .initial = std::chrono::milliseconds{1}, .max = std::chrono::milliseconds{1}
+                }
+            ),
+            std::runtime_error
+        );
+    });
+}
+
+TEST_F(BackendInterfaceRetryCoroTest, RetryOnTimeoutCoroDoesNotBlockItsThread)
+{
+    bool ran = false;
+
+    runSpawn([&ran, this](auto yield) {
+        boost::asio::post(ctx_, [&ran]() { ran = true; });
+
+        std::size_t calls = 0;
+        retryOnTimeout(
+            [&calls]() -> int {
+                if (++calls < 2)
+                    throw DatabaseError{};
+                return 0;
+            },
+            yield,
+            util::Retry::Delays{
+                .initial = std::chrono::milliseconds{20}, .max = std::chrono::milliseconds{20}
+            }
+        );
+
+        EXPECT_TRUE(ran);
+    });
+}
diff --git a/tests/unit/data/cassandra/ErrorTests.cpp b/tests/unit/data/cassandra/ErrorTests.cpp
new file mode 100644
index 000000000..5739a156e
--- /dev/null
+++ b/tests/unit/data/cassandra/ErrorTests.cpp
@@ -0,0 +1,43 @@
+#include "data/cassandra/Error.hpp"
+
+#include <cassandra.h>
+#include <gtest/gtest.h>
+
+#include <cstdint>
+
+using namespace data::cassandra;
+
+namespace {
+
+CassandraError
+makeError(uint32_t const code)
+{
+    return CassandraError{"some error", code};
+}
+
+}  // namespace
+
+// isInvalidQuery is load bearing: DefaultExecutionStrategy::throwErrorIfNeeded treats it as the
+// only permanent failure and retries everything else, so anything wrongly reported here would be
+// retried forever (if false) or surfaced as a fatal std::runtime_error (if true).
+TEST(BackendCassandraErrorTest, IsInvalidQueryOnlyForInvalidQuery)
+{
+    EXPECT_TRUE(makeError(CASS_ERROR_SERVER_INVALID_QUERY).isInvalidQuery());
+
+    EXPECT_FALSE(makeError(CASS_ERROR_SERVER_READ_FAILURE).isInvalidQuery());
+    EXPECT_FALSE(makeError(CASS_ERROR_SERVER_WRITE_FAILURE).isInvalidQuery());
+    EXPECT_FALSE(makeError(CASS_ERROR_SERVER_READ_TIMEOUT).isInvalidQuery());
+    EXPECT_FALSE(makeError(CASS_ERROR_SERVER_UNAVAILABLE).isInvalidQuery());
+    EXPECT_FALSE(makeError(CASS_ERROR_SERVER_SYNTAX_ERROR).isInvalidQuery());
+    EXPECT_FALSE(makeError(CASS_OK).isInvalidQuery());
+}
+
+TEST(BackendCassandraErrorTest, MessageAndCodeArePreserved)
+{
+    // throwErrorIfNeeded puts message() into the DatabaseError it throws, so that treating an
+    // unclassified error as a timeout still reports what actually failed
+    auto const err = CassandraError{"received 1 responses and 1 failures", 0x1300};
+
+    EXPECT_EQ(err.message(), "received 1 responses and 1 failures");
+    EXPECT_EQ(err.code(), 0x1300u);
+}
diff --git a/tests/unit/data/cassandra/ExecutionStrategyTests.cpp b/tests/unit/data/cassandra/ExecutionStrategyTests.cpp
index 7a8efc633..a16cab462 100644
--- a/tests/unit/data/cassandra/ExecutionStrategyTests.cpp
+++ b/tests/unit/data/cassandra/ExecutionStrategyTests.cpp
@@ -19,6 +19,7 @@
 #include <memory>
 #include <optional>
 #include <stdexcept>
+#include <string>
 #include <thread>
 #include <vector>
 
@@ -154,7 +155,45 @@ TEST_F(BackendCassandraExecutionStrategyTest, ReadOneInCoroutineThrowsOnTimeoutF
 
     runSpawn([&strat](boost::asio::yield_context yield) {
         auto statement = FakeStatement{};
-        EXPECT_THROW(strat.read(yield, statement), data::DatabaseTimeout);
+        EXPECT_THROW(strat.read(yield, statement), data::DatabaseError);
+    });
+}
+
+// A CL=QUORUM read failure is CASS_ERROR_SERVER_READ_FAILURE, which used to not throw at all,
+// leaving read() spinning; Times(1) pins that down.
+TEST_F(BackendCassandraExecutionStrategyTest, ReadOneInCoroutineThrowsOnQuorumReadFailure)
+{
+    auto strat = makeStrategy();
+
+    ON_CALL(
+        handle_,
+        asyncExecute(A<FakeStatement const&>(), A<std::function<void(FakeResultOrError)>&&>())
+    )
+        .WillByDefault([](auto const&, auto&& cb) {
+            auto res = FakeResultOrError{CassandraError{
+                "received 1 responses and 1 failures", CASS_ERROR_SERVER_READ_FAILURE
+            }};
+            cb(res);  // notify that item is ready
+            return FakeFutureWithCallback{res};
+        });
+    EXPECT_CALL(
+        handle_,
+        asyncExecute(A<FakeStatement const&>(), A<std::function<void(FakeResultOrError)>&&>())
+    )
+        .Times(1);
+    EXPECT_CALL(*counters_, registerReadStartedImpl(1));
+    EXPECT_CALL(*counters_, registerReadErrorImpl(1));
+
+    runSpawn([&strat](boost::asio::yield_context yield) {
+        auto statement = FakeStatement{};
+        try {
+            strat.read(yield, statement);
+            FAIL() << "expected DatabaseError";
+        } catch (data::DatabaseError const& e) {
+            EXPECT_THAT(
+                std::string{e.what()}, testing::HasSubstr("received 1 responses and 1 failures")
+            );
+        }
     });
 }
 
@@ -246,7 +285,7 @@ TEST_F(BackendCassandraExecutionStrategyTest, ReadBatchInCoroutineThrowsOnTimeou
 
     runSpawn([&strat](boost::asio::yield_context yield) {
         auto statements = std::vector<FakeStatement>(kNumStatements);
-        EXPECT_THROW(strat.read(yield, statements), data::DatabaseTimeout);
+        EXPECT_THROW(strat.read(yield, statements), data::DatabaseError);
     });
 }
 
@@ -384,7 +423,7 @@ TEST_F(BackendCassandraExecutionStrategyTest, ReadEachInCoroutineThrowsOnFailure
 
     runSpawn([&strat](boost::asio::yield_context yield) {
         auto statements = std::vector<FakeStatement>(kNumStatements);
-        EXPECT_THROW(strat.readEach(yield, statements), data::DatabaseTimeout);
+        EXPECT_THROW(strat.readEach(yield, statements), data::DatabaseError);
     });
 }
 
diff --git a/tests/unit/data/cassandra/SettingsProviderTests.cpp b/tests/unit/data/cassandra/SettingsProviderTests.cpp
index 51919611b..ea27d6401 100644
--- a/tests/unit/data/cassandra/SettingsProviderTests.cpp
+++ b/tests/unit/data/cassandra/SettingsProviderTests.cpp
@@ -47,7 +47,11 @@ getParseSettingsConfig(boost::json::value val)
         {"database.cassandra.write_batch_size", ConfigValue{ConfigType::Integer}.defaultValue(20)},
         {"database.cassandra.connect_timeout", ConfigValue{ConfigType::Integer}.optional()},
         {"database.cassandra.certfile", ConfigValue{ConfigType::String}.optional()},
-        {"database.cassandra.request_timeout", ConfigValue{ConfigType::Integer}.defaultValue(0)},
+        {"database.cassandra.initial_request_retry_delay",
+         ConfigValue{ConfigType::Double}.defaultValue(0.5)},
+        {"database.cassandra.max_request_retry_delay",
+         ConfigValue{ConfigType::Double}.defaultValue(5.0)},
+        {"database.cassandra.request_timeout", ConfigValue{ConfigType::Double}.optional()},
         {"database.cassandra.secure_connect_bundle", ConfigValue{ConfigType::String}.optional()},
         {"database.cassandra.username", ConfigValue{ConfigType::String}.optional()},
         {"database.cassandra.password", ConfigValue{ConfigType::String}.optional()},
@@ -163,3 +167,72 @@ TEST_F(SettingsProviderTest, CertificateConfig)
     auto const settings = provider.getSettings();
     EXPECT_EQ(settings.certificate, "certificateData");
 }
+
+TEST_F(SettingsProviderTest, RequestTimeoutAcceptsFractionalSeconds)
+{
+    auto const cfg = getParseSettingsConfig(
+        boost::json::parse(
+            R"JSON({"database.cassandra.contact_points": "127.0.0.1",
+                "database.cassandra.request_timeout": 2.5})JSON"
+        )
+    );
+    SettingsProvider const provider{cfg.getObject("database.cassandra")};
+
+    EXPECT_EQ(provider.getSettings().requestTimeout, std::chrono::milliseconds{2500});
+}
+
+TEST_F(SettingsProviderTest, RequestTimeoutStillAcceptsWholeSeconds)
+{
+    // a JSON integer must still be accepted
+    auto const cfg = getParseSettingsConfig(
+        boost::json::parse(
+            R"JSON({"database.cassandra.contact_points": "127.0.0.1",
+                "database.cassandra.request_timeout": 3})JSON"
+        )
+    );
+    SettingsProvider const provider{cfg.getObject("database.cassandra")};
+
+    EXPECT_EQ(provider.getSettings().requestTimeout, std::chrono::milliseconds{3000});
+}
+
+TEST_F(SettingsProviderTest, RetryDelaysDefaults)
+{
+    auto const cfg = getParseSettingsConfig(
+        boost::json::parse(R"JSON({"database.cassandra.contact_points": "127.0.0.1"})JSON")
+    );
+    SettingsProvider const provider{cfg.getObject("database.cassandra")};
+
+    EXPECT_EQ(provider.getInitialRetryDelay(), std::chrono::milliseconds{500});
+    EXPECT_EQ(provider.getMaxRetryDelay(), std::chrono::milliseconds{5000});
+}
+
+TEST_F(SettingsProviderTest, RetryDelaysAreIndependentOfRequestTimeout)
+{
+    auto const cfg = getParseSettingsConfig(
+        boost::json::parse(
+            R"JSON({"database.cassandra.contact_points": "127.0.0.1",
+                "database.cassandra.request_timeout": 2.5,
+                "database.cassandra.initial_request_retry_delay": 0.25,
+                "database.cassandra.max_request_retry_delay": 1.5})JSON"
+        )
+    );
+    SettingsProvider const provider{cfg.getObject("database.cassandra")};
+
+    EXPECT_EQ(provider.getSettings().requestTimeout, std::chrono::milliseconds{2500});
+    EXPECT_EQ(provider.getInitialRetryDelay(), std::chrono::milliseconds{250});
+    EXPECT_EQ(provider.getMaxRetryDelay(), std::chrono::milliseconds{1500});
+}
+
+TEST_F(SettingsProviderTest, EqualRetryDelaysDisableBackoff)
+{
+    auto const cfg = getParseSettingsConfig(
+        boost::json::parse(
+            R"JSON({"database.cassandra.contact_points": "127.0.0.1",
+                "database.cassandra.initial_request_retry_delay": 0.5,
+                "database.cassandra.max_request_retry_delay": 0.5})JSON"
+        )
+    );
+    SettingsProvider const provider{cfg.getObject("database.cassandra")};
+
+    EXPECT_EQ(provider.getInitialRetryDelay(), provider.getMaxRetryDelay());
+}
diff --git a/tests/unit/rpc/RPCEngineTests.cpp b/tests/unit/rpc/RPCEngineTests.cpp
index 57ffaa0d8..48820e48b 100644
--- a/tests/unit/rpc/RPCEngineTests.cpp
+++ b/tests/unit/rpc/RPCEngineTests.cpp
@@ -262,7 +262,7 @@ TEST_F(RPCEngineTest, ThrowDatabaseError)
     EXPECT_CALL(*backend_, isTooBusy).WillOnce(Return(false));
     EXPECT_CALL(*handlerProvider, getHandler(method))
         .WillOnce(Return(AnyHandler{tests::common::FailingHandlerFake{}}));
-    EXPECT_CALL(*mockCountersPtr_, rpcErrored(method)).WillOnce(Throw(data::DatabaseTimeout{}));
+    EXPECT_CALL(*mockCountersPtr_, rpcErrored(method)).WillOnce(Throw(data::DatabaseError{}));
     EXPECT_CALL(*handlerProvider, contains(method)).WillOnce(Return(true));
     EXPECT_CALL(*mockCountersPtr_, onTooBusy());
 
diff --git a/tests/unit/util/RetryTests.cpp b/tests/unit/util/RetryTests.cpp
index 3a2d786e4..f6cbf799c 100644
--- a/tests/unit/util/RetryTests.cpp
+++ b/tests/unit/util/RetryTests.cpp
@@ -1,6 +1,7 @@
 #include "util/AsioContextTestFixture.hpp"
 #include "util/Retry.hpp"
 
+#include <boost/asio/post.hpp>
 #include <boost/asio/strand.hpp>
 #include <gmock/gmock.h>
 #include <gtest/gtest.h>
@@ -18,7 +19,7 @@ protected:
 
 TEST_F(RetryTests, ExponentialBackoffStrategy)
 {
-    ExponentialBackoffStrategy strategy{delay_, maxDelay_};
+    ExponentialBackoffStrategy strategy{{.initial = delay_, .max = maxDelay_}};
 
     EXPECT_EQ(strategy.getDelay(), delay_);
 
@@ -46,7 +47,10 @@ struct RetryWithExponentialBackoffStrategyTests : SyncAsioContextTest, RetryTest
     }
 
 protected:
-    Retry retry_ = makeRetryExponentialBackoff(delay_, maxDelay_, boost::asio::make_strand(ctx_));
+    Retry retry_ = makeRetryExponentialBackoff(
+        {.initial = delay_, .max = maxDelay_},
+        boost::asio::make_strand(ctx_)
+    );
     testing::MockFunction<void()> mockCallback_;
 };
 
@@ -88,3 +92,44 @@ TEST_F(RetryWithExponentialBackoffStrategyTests, Reset)
     EXPECT_EQ(retry_.attemptNumber(), 0);
     EXPECT_EQ(retry_.delayValue(), delay_);
 }
+
+struct RetryWaitTests : SyncAsioContextTest, RetryTests {};
+
+TEST_F(RetryWaitTests, WaitOnCoroutineAdvancesAttemptAndDelay)
+{
+    runSpawn([this](auto yield) {
+        auto retry = makeRetryExponentialBackoff(
+            {.initial = delay_, .max = maxDelay_}, yield.get_executor()
+        );
+
+        EXPECT_EQ(retry.attemptNumber(), 0);
+        EXPECT_EQ(retry.delayValue(), delay_);
+
+        retry.wait(yield);
+        EXPECT_EQ(retry.attemptNumber(), 1);
+        EXPECT_EQ(retry.delayValue(), delay_ * 2);
+
+        retry.wait(yield);
+        EXPECT_EQ(retry.attemptNumber(), 2);
+    });
+}
+
+TEST_F(RetryWaitTests, WaitDoesNotBlockItsThread)
+{
+    bool ran = false;
+    bool ranBeforeWaitReturned = false;
+
+    runSpawn([this, &ran, &ranBeforeWaitReturned](auto yield) {
+        boost::asio::post(ctx_, [&ran]() { ran = true; });
+
+        auto retry = makeRetryExponentialBackoff(
+            {.initial = std::chrono::milliseconds{20}, .max = std::chrono::milliseconds{20}},
+            yield.get_executor()
+        );
+        retry.wait(yield);
+
+        ranBeforeWaitReturned = ran;
+    });
+
+    EXPECT_TRUE(ranBeforeWaitReturned);
+}