From 4aaca6540f1192fff6f7b9d1b7d8fbb2a826cd4e Mon Sep 17 00:00:00 2001
From: Steve Boyd <emteknetnz@gmail.com>
Date: Thu, 3 Oct 2024 10:41:10 +1300
Subject: [PATCH] NEW Allow database read-only replicas

---
 bin/sake                                      |   4 +
 src/Control/Director.php                      |  21 ++
 src/Core/ClassInfo.php                        |   2 +-
 src/Core/CoreKernel.php                       | 140 ++++++++--
 src/Dev/SapphireTest.php                      |   5 +-
 src/ORM/Connect/TempDatabase.php              |   2 +-
 src/ORM/DB.php                                | 231 ++++++++++++++--
 src/ORM/DataList.php                          |   4 +-
 src/ORM/DataObject.php                        |   8 +
 src/ORM/DataQuery.php                         |  41 ++-
 src/Security/Group.php                        |   2 +
 src/Security/LoginAttempt.php                 |   2 +
 src/Security/Member.php                       |   2 +
 src/Security/MemberPassword.php               |   2 +
 src/Security/Permission.php                   |  80 +++---
 src/Security/PermissionRole.php               |   2 +
 src/Security/PermissionRoleCode.php           |   2 +
 src/Security/RememberLoginHash.php            |   2 +
 src/Security/Security.php                     |   6 +
 tests/php/Core/KernelTest.php                 |  32 +++
 tests/php/ORM/DBReplicaTest.php               | 260 ++++++++++++++++++
 tests/php/ORM/DBReplicaTest.yml               |  18 ++
 .../php/ORM/DBReplicaTest/TestController.php  |  18 ++
 tests/php/ORM/DBReplicaTest/TestObject.php    |  16 ++
 tests/php/ORM/MySQLiConnectorTest.php         |   2 +-
 25 files changed, 801 insertions(+), 103 deletions(-)
 create mode 100644 tests/php/ORM/DBReplicaTest.php
 create mode 100644 tests/php/ORM/DBReplicaTest.yml
 create mode 100644 tests/php/ORM/DBReplicaTest/TestController.php
 create mode 100644 tests/php/ORM/DBReplicaTest/TestObject.php

diff --git a/bin/sake b/bin/sake
index 713ab522689..8fcc0f0f0df 100755
--- a/bin/sake
+++ b/bin/sake
@@ -2,6 +2,7 @@
 <?php
 
 use SilverStripe\Cli\Sake;
+use SilverStripe\ORM\DB;
 
 // Ensure that people can't access this from a web-server
 if (!in_array(PHP_SAPI, ['cli', 'cgi', 'cgi-fcgi'])) {
@@ -11,5 +12,8 @@ if (!in_array(PHP_SAPI, ['cli', 'cgi', 'cgi-fcgi'])) {
 
 require_once __DIR__ . '/../src/includes/autoload.php';
 
+// CLI scripts must only use the primary database connection and not replicas
+DB::setMustUsePrimary();
+
 $sake = new Sake();
 $sake->run();
diff --git a/src/Control/Director.php b/src/Control/Director.php
index 3ff9a456c7a..f1f4b7fe5f3 100644
--- a/src/Control/Director.php
+++ b/src/Control/Director.php
@@ -17,6 +17,7 @@
 use SilverStripe\View\Requirements;
 use SilverStripe\View\Requirements_Backend;
 use SilverStripe\View\TemplateGlobalProvider;
+use SilverStripe\ORM\DB;
 
 /**
  * Director is responsible for processing URLs, and providing environment information.
@@ -84,6 +85,14 @@ class Director implements TemplateGlobalProvider
      */
     private static $default_base_url = '`SS_BASE_URL`';
 
+    /**
+     * List of routing rule patterns that must only use the primary database and not a replica
+     */
+    private static array $rule_patterns_must_use_primary_db = [
+        'dev',
+        'Security',
+    ];
+
     public function __construct()
     {
     }
@@ -296,6 +305,18 @@ public function handleRequest(HTTPRequest $request)
     {
         Injector::inst()->registerService($request, HTTPRequest::class);
 
+        // Check if primary database must be used based on request rules
+        // Note this check must happend before the rules are processed as
+        // $shiftOnSuccess param is passed as true in `$request->match($pattern, true)` later on in
+        // this method, which modifies `$this->dirParts`, thus affecting `$request->match($rule)` directly below
+        $primaryDbOnlyRules = Director::config()->uninherited('rule_patterns_must_use_primary_db');
+        foreach ($primaryDbOnlyRules as $rule) {
+            if ($request->match($rule)) {
+                DB::setMustUsePrimary();
+                break;
+            }
+        }
+
         $rules = Director::config()->uninherited('rules');
 
         $this->extend('updateRules', $rules);
diff --git a/src/Core/ClassInfo.php b/src/Core/ClassInfo.php
index fd32d9cb545..bd2ab56d6f7 100644
--- a/src/Core/ClassInfo.php
+++ b/src/Core/ClassInfo.php
@@ -85,7 +85,7 @@ public static function exists($class)
     public static function hasTable($tableName)
     {
         $cache = ClassInfo::getCache();
-        $configData = serialize(DB::getConfig());
+        $configData = serialize(DB::getConfig(DB::CONN_PRIMARY));
         $cacheKey = 'tableList_' . md5($configData);
         $tableList = $cache->get($cacheKey) ?? [];
         if (empty($tableList) && DB::is_active()) {
diff --git a/src/Core/CoreKernel.php b/src/Core/CoreKernel.php
index e3f92b8f7a7..ca9946f68a3 100644
--- a/src/Core/CoreKernel.php
+++ b/src/Core/CoreKernel.php
@@ -7,12 +7,14 @@
 use SilverStripe\ORM\Connect\NullDatabase;
 use SilverStripe\ORM\DB;
 use Exception;
+use InvalidArgumentException;
 
 /**
  * Simple Kernel container
  */
 class CoreKernel extends BaseKernel
 {
+
     protected bool $bootDatabase = true;
 
     /**
@@ -38,7 +40,7 @@ public function boot($flush = false)
         $this->flush = $flush;
 
         if (!$this->bootDatabase) {
-            DB::set_conn(new NullDatabase());
+            DB::set_conn(new NullDatabase(), DB::CONN_PRIMARY);
         }
 
         $this->bootPHP();
@@ -73,7 +75,7 @@ protected function validateDatabase()
     }
 
     /**
-     * Load default database configuration from the $database and $databaseConfig globals
+     * Load database configuration from the $database and $databaseConfig globals
      */
     protected function bootDatabaseGlobals()
     {
@@ -84,41 +86,62 @@ protected function bootDatabaseGlobals()
         global $databaseConfig;
         global $database;
 
-        // Case 1: $databaseConfig global exists. Merge $database in as needed
-        if (!empty($databaseConfig)) {
-            if (!empty($database)) {
-                $databaseConfig['database'] =  $this->getDatabasePrefix() . $database . $this->getDatabaseSuffix();
+        // Ensure global database config has prefix and suffix applied
+        if (!empty($databaseConfig) && !empty($database)) {
+            $databaseConfig['database'] = $this->getDatabasePrefix() . $database . $this->getDatabaseSuffix();
+        }
+
+        // Set config for primary and any replicas
+        for ($i = 0; $i <= DB::MAX_REPLICAS; $i++) {
+            if ($i === 0) {
+                $name = DB::CONN_PRIMARY;
+            } else {
+                $name = DB::getReplicaConfigKey($i);
+                if (!DB::hasConfig($name)) {
+                    break;
+                }
             }
 
+            // Case 1: $databaseConfig global exists
             // Only set it if its valid, otherwise ignore $databaseConfig entirely
-            if (!empty($databaseConfig['database'])) {
-                DB::setConfig($databaseConfig);
-
+            if (!empty($databaseConfig) && !empty($databaseConfig['database'])) {
+                DB::setConfig($databaseConfig, $name);
                 return;
             }
-        }
 
-        // Case 2: $database merged into existing config
-        if (!empty($database)) {
-            $existing = DB::getConfig();
-            $existing['database'] = $this->getDatabasePrefix() . $database . $this->getDatabaseSuffix();
-
-            DB::setConfig($existing);
+            // Case 2: $databaseConfig global does not exist
+            // Merge $database global into existing config
+            if (!empty($database)) {
+                $dbConfig = DB::getConfig($name);
+                $dbConfig['database'] = $this->getDatabasePrefix() . $database . $this->getDatabaseSuffix();
+                DB::setConfig($dbConfig, $name);
+            }
         }
     }
 
     /**
-     * Load default database configuration from environment variable
+     * Load database configuration from environment variables
      */
     protected function bootDatabaseEnvVars()
     {
         if (!$this->bootDatabase) {
             return;
         }
-        // Set default database config
+        // Set primary database config
         $databaseConfig = $this->getDatabaseConfig();
         $databaseConfig['database'] = $this->getDatabaseName();
-        DB::setConfig($databaseConfig);
+        DB::setConfig($databaseConfig, DB::CONN_PRIMARY);
+
+        // Set database replicas config
+        for ($i = 1; $i <= DB::MAX_REPLICAS; $i++) {
+            $envKey = $this->getReplicaEnvKey('SS_DATABASE_SERVER', $i);
+            if (!Environment::hasEnv($envKey)) {
+                break;
+            }
+            $replicaDatabaseConfig = $this->getDatabaseReplicaConfig($i);
+            $configKey = DB::getReplicaConfigKey($i);
+            DB::setConfig($replicaDatabaseConfig, $configKey);
+        }
     }
 
     /**
@@ -127,12 +150,72 @@ protected function bootDatabaseEnvVars()
      * @return array
      */
     protected function getDatabaseConfig()
+    {
+        return $this->getSingleDataBaseConfig(0);
+    }
+
+    private function getDatabaseReplicaConfig(int $replica)
+    {
+        if ($replica <= 0) {
+            throw new InvalidArgumentException('Replica number must be greater than 0');
+        }
+        return $this->getSingleDataBaseConfig($replica);
+    }
+
+    /**
+     * Convert a database key to a replica key
+     * e.g. SS_DATABASE_SERVER -> SS_DATABASE_SERVER_REPLICA_01
+     *
+     * @param string $key - The key to look up in the environment
+     * @param int $replica - Replica number
+     */
+    private function getReplicaEnvKey(string $key, int $replica): string
+    {
+        if ($replica <= 0) {
+            throw new InvalidArgumentException('Replica number must be greater than 0');
+        }
+        // Do not allow replicas to define keys that could lead to unexpected behaviour if
+        // they do not match the primary database configuration
+        if (in_array($key, ['SS_DATABASE_CLASS', 'SS_DATABASE_NAME', 'SS_DATABASE_CHOOSE_NAME'])) {
+            return $key;
+        }
+        // Left pad replica number with a zeros to match the length of the maximum replica number
+        $len = strlen((string) DB::MAX_REPLICAS);
+        return $key . '_REPLICA_' . str_pad($replica, $len, '0', STR_PAD_LEFT);
+    }
+
+    /**
+     * Reads a single database configuration variable from the environment
+     * For replica databases, it will first attempt to find replica-specific configuration
+     * before falling back to the primary configuration.
+     *
+     * Replicate specific configuration has `_REPLICA_01` appended to the key
+     * where 01 is the replica number.
+     *
+     * @param string $key - The key to look up in the environment
+     * @param int $replica - Replica number. Passing 0 will return the primary database configuration
+     */
+    private function getDatabaseConfigVariable(string $key, int $replica): string
+    {
+        if ($replica > 0) {
+            $key = $this->getReplicaEnvKey($key, $replica);
+        }
+        if (Environment::hasEnv($key)) {
+            return Environment::getEnv($key);
+        }
+        return '';
+    }
+
+    /**
+     * @param int $replica - Replica number. Passing 0 will return the primary database configuration
+     */
+    private function getSingleDataBaseConfig(int $replica): array
     {
         $databaseConfig = [
-            "type" => Environment::getEnv('SS_DATABASE_CLASS') ?: 'MySQLDatabase',
-            "server" => Environment::getEnv('SS_DATABASE_SERVER') ?: 'localhost',
-            "username" => Environment::getEnv('SS_DATABASE_USERNAME') ?: null,
-            "password" => Environment::getEnv('SS_DATABASE_PASSWORD') ?: null,
+            "type" => $this->getDatabaseConfigVariable('SS_DATABASE_CLASS', $replica) ?: 'MySQLDatabase',
+            "server" => $this->getDatabaseConfigVariable('SS_DATABASE_SERVER', $replica) ?: 'localhost',
+            "username" => $this->getDatabaseConfigVariable('SS_DATABASE_USERNAME', $replica) ?: null,
+            "password" => $this->getDatabaseConfigVariable('SS_DATABASE_PASSWORD', $replica) ?: null,
         ];
 
         // Only add SSL keys in the array if there is an actual value associated with them
@@ -143,7 +226,7 @@ protected function getDatabaseConfig()
             'ssl_cipher' => 'SS_DATABASE_SSL_CIPHER',
         ];
         foreach ($sslConf as $key => $envVar) {
-            $envValue = Environment::getEnv($envVar);
+            $envValue = $this->getDatabaseConfigVariable($envVar, $replica);
             if ($envValue) {
                 $databaseConfig[$key] = $envValue;
             }
@@ -159,25 +242,25 @@ protected function getDatabaseConfig()
         }
 
         // Set the port if called for
-        $dbPort = Environment::getEnv('SS_DATABASE_PORT');
+        $dbPort = $this->getDatabaseConfigVariable('SS_DATABASE_PORT', $replica);
         if ($dbPort) {
             $databaseConfig['port'] = $dbPort;
         }
 
         // Set the timezone if called for
-        $dbTZ = Environment::getEnv('SS_DATABASE_TIMEZONE');
+        $dbTZ = $this->getDatabaseConfigVariable('SS_DATABASE_TIMEZONE', $replica);
         if ($dbTZ) {
             $databaseConfig['timezone'] = $dbTZ;
         }
 
         // For schema enabled drivers:
-        $dbSchema = Environment::getEnv('SS_DATABASE_SCHEMA');
+        $dbSchema = $this->getDatabaseConfigVariable('SS_DATABASE_SCHEMA', $replica);
         if ($dbSchema) {
             $databaseConfig["schema"] = $dbSchema;
         }
 
         // For SQlite3 memory databases (mainly for testing purposes)
-        $dbMemory = Environment::getEnv('SS_DATABASE_MEMORY');
+        $dbMemory = $this->getDatabaseConfigVariable('SS_DATABASE_MEMORY', $replica);
         if ($dbMemory) {
             $databaseConfig["memory"] = $dbMemory;
         }
@@ -205,6 +288,7 @@ protected function getDatabaseSuffix()
 
     /**
      * Get name of database
+     * Note that any replicas must have the same database name as the primary database
      *
      * @return string
      */
diff --git a/src/Dev/SapphireTest.php b/src/Dev/SapphireTest.php
index 10025a22d63..d735290d482 100644
--- a/src/Dev/SapphireTest.php
+++ b/src/Dev/SapphireTest.php
@@ -48,7 +48,7 @@
 use SilverStripe\Dev\Exceptions\ExpectedErrorException;
 use SilverStripe\Dev\Exceptions\ExpectedNoticeException;
 use SilverStripe\Dev\Exceptions\ExpectedWarningException;
-use SilverStripe\Dev\Exceptions\UnexpectedErrorException;
+use SilverStripe\ORM\DB;
 
 /**
  * Test case class for the Silverstripe framework.
@@ -434,6 +434,9 @@ protected function currentTestDisablesDatabase()
      */
     public static function setUpBeforeClass(): void
     {
+        // Disallow the use of DB replicas in tests
+        DB::setMustUsePrimary();
+
         // Start tests
         static::start();
 
diff --git a/src/ORM/Connect/TempDatabase.php b/src/ORM/Connect/TempDatabase.php
index 598339ea22e..66dae3e91c8 100644
--- a/src/ORM/Connect/TempDatabase.php
+++ b/src/ORM/Connect/TempDatabase.php
@@ -43,7 +43,7 @@ class TempDatabase
      *
      * @param string $name DB Connection name to use
      */
-    public function __construct($name = 'default')
+    public function __construct($name = DB::CONN_PRIMARY)
     {
         $this->name = $name;
     }
diff --git a/src/ORM/DB.php b/src/ORM/DB.php
index 054b858d02f..4846769b356 100644
--- a/src/ORM/DB.php
+++ b/src/ORM/DB.php
@@ -3,6 +3,7 @@
 namespace SilverStripe\ORM;
 
 use InvalidArgumentException;
+use RunTimeException;
 use SilverStripe\Control\Director;
 use SilverStripe\Control\HTTPRequest;
 use SilverStripe\Core\Config\Config;
@@ -21,6 +22,22 @@
  */
 class DB
 {
+    /**
+     * A dynamic connection that will use either a replica connection if one is
+     * available and not forced to use the 'primary' connection, or the 'primary' connection
+     */
+    public const CONN_DYNAMIC = 'dynamic';
+
+    /**
+     * The 'primary' connection name, which is the main database connection and is used for all write
+     * operations and for read operations when the 'default' connection is forced to use the 'primary' connection
+     */
+    public const CONN_PRIMARY = 'primary';
+
+    /**
+     * The maximum number of replicas databases that can be configured
+     */
+    public const MAX_REPLICAS = 99;
 
     /**
      * This constant was added in SilverStripe 2.4 to indicate that SQL-queries
@@ -58,19 +75,45 @@ class DB
      */
     protected static $configs = [];
 
-
-
     /**
      * The last SQL query run.
      * @var string
      */
     public static $lastQuery;
 
+    /**
+     * The name of the last connection used. This is only used for unit-testing purposes.
+     * @interal
+     */
+    private static string $lastConnectionName = '';
+
     /**
      * Internal flag to keep track of when db connection was attempted.
      */
     private static $connection_attempted = false;
 
+    /**
+     * Only use the primary database connection for the rest of the current request
+     *
+     * @internal
+     */
+    private static bool $mustUsePrimary = false;
+
+    /**
+     * Used by DB::withPrimary() to count the number of times it has been called
+     * Uses an int instead of a bool to allow for nested calls
+     *
+     * @internal
+     */
+    private static int $withPrimaryCount = 0;
+
+    /**
+     * The key of the replica config to use for this request
+     *
+     * @internal
+     */
+    private static string $replicaConfigKey = '';
+
     /**
      * Set the global database connection.
      * Pass an object that's a subclass of SS_Database.  This object will be used when {@link DB::query()}
@@ -78,11 +121,11 @@ class DB
      *
      * @param Database $connection The connection object to set as the connection.
      * @param string $name The name to give to this connection.  If you omit this argument, the connection
-     * will be the default one used by the ORM.  However, you can store other named connections to
+     * will be the primary one used by the ORM.  However, you can store other named connections to
      * be accessed through DB::get_conn($name).  This is useful when you have an application that
      * needs to connect to more than one database.
      */
-    public static function set_conn(Database $connection, $name = 'default')
+    public static function set_conn(Database $connection, $name = DB::CONN_PRIMARY)
     {
         DB::$connections[$name] = $connection;
     }
@@ -92,11 +135,17 @@ public static function set_conn(Database $connection, $name = 'default')
      *
      * @param string $name An optional name given to a connection in the DB::setConn() call.  If omitted,
      * the default connection is returned.
-     * @return Database
+     * @return Database|null
      */
-    public static function get_conn($name = 'default')
+    public static function get_conn($name = DB::CONN_DYNAMIC)
     {
+        // Allow default to connect to replica if configured
+        if ($name === DB::CONN_DYNAMIC) {
+            $name = DB::getDynamicConnectionName();
+        }
+
         if (isset(DB::$connections[$name])) {
+            DB::$lastConnectionName = $name;
             return DB::$connections[$name];
         }
 
@@ -109,14 +158,50 @@ public static function get_conn($name = 'default')
         return null;
     }
 
+    /**
+     * Whether the primary database connection will be used if the database is used right now
+     */
+    public static function willUsePrimary(): bool
+    {
+        return DB::$mustUsePrimary || DB::$withPrimaryCount > 0 || !DB::hasReplicaConfig();
+    }
+
+    /**
+     * Set to use the primary database connection for rest of the current request
+     * meaning that replia connections will no longer be used
+     *
+     * This intentioally does not have a parameter to set this back to false, as this it to prevent
+     * accidentally attempting writing to a replica, or reading from an out of date replica
+     * after a write
+     */
+    public static function setMustUsePrimary(): void
+    {
+        DB::$mustUsePrimary = true;
+    }
+
+    /**
+     * Only use the primary database connection when calling $callback
+     * Use this when doing non-mutable queries on the primary database where querying
+     * an out of sync replica could cause issues
+     * There's no need to use this with mutable queries, or after calling a mutable query
+     * as the primary database connection will be automatically used
+     */
+    public static function withPrimary(callable $callback): mixed
+    {
+        DB::$withPrimaryCount++;
+        $result = $callback();
+        DB::$withPrimaryCount--;
+        return $result;
+    }
+
     /**
      * Retrieves the schema manager for the current database
      *
-     * @param string $name An optional name given to a connection in the DB::setConn() call.  If omitted,
-     * the default connection is returned.
-     * @return DBSchemaManager
+     * @param string $name An optional name given to a connection in the DB::setConn() call.
+     * If omitted, a dynamic connection is returned.
+     * @return DBSchemaManager|null
      */
-    public static function get_schema($name = 'default')
+    public static function get_schema($name = DB::CONN_DYNAMIC)
     {
         $connection = DB::get_conn($name);
         if ($connection) {
@@ -130,11 +215,11 @@ public static function get_schema($name = 'default')
      *
      * @param SQLExpression $expression The expression object to build from
      * @param array $parameters Out parameter for the resulting query parameters
-     * @param string $name An optional name given to a connection in the DB::setConn() call.  If omitted,
-     * the default connection is returned.
-     * @return string The resulting SQL as a string
+     * @param string $name An optional name given to a connection in the DB::setConn() call.
+     * If omitted, a dynamic connection is returned.
+     * @return string|null The resulting SQL as a string
      */
-    public static function build_sql(SQLExpression $expression, &$parameters, $name = 'default')
+    public static function build_sql(SQLExpression $expression, &$parameters, $name = DB::CONN_DYNAMIC)
     {
         $connection = DB::get_conn($name);
         if ($connection) {
@@ -148,11 +233,11 @@ public static function build_sql(SQLExpression $expression, &$parameters, $name
     /**
      * Retrieves the connector object for the current database
      *
-     * @param string $name An optional name given to a connection in the DB::setConn() call.  If omitted,
-     * the default connection is returned.
-     * @return DBConnector
+     * @param string $name An optional name given to a connection in the DB::setConn() call.
+     * If omitted, a dynamic connection is returned.
+     * @return DBConnector|null
      */
-    public static function get_connector($name = 'default')
+    public static function get_connector($name = DB::CONN_DYNAMIC)
     {
         $connection = DB::get_conn($name);
         if ($connection) {
@@ -268,8 +353,13 @@ public static function valid_alternative_database_name($name)
      * @param string $label identifier for the connection
      * @return Database
      */
-    public static function connect($databaseConfig, $label = 'default')
+    public static function connect($databaseConfig, $label = DB::CONN_DYNAMIC)
     {
+        // Allow default to connect to replica if configured
+        if ($label === DB::CONN_DYNAMIC) {
+            $label = DB::getDynamicConnectionName();
+        }
+
         // This is used by the "testsession" module to test up a test session using an alternative name
         if ($name = DB::get_alternative_database_name()) {
             $databaseConfig['database'] = $name;
@@ -288,6 +378,7 @@ public static function connect($databaseConfig, $label = 'default')
         $conn = Injector::inst()->create($dbClass);
         DB::set_conn($conn, $label);
         $conn->connect($databaseConfig);
+        DB::$lastConnectionName = $label;
 
         return $conn;
     }
@@ -298,7 +389,7 @@ public static function connect($databaseConfig, $label = 'default')
      * @param array $databaseConfig
      * @param string $name
      */
-    public static function setConfig($databaseConfig, $name = 'default')
+    public static function setConfig($databaseConfig, $name = DB::CONN_PRIMARY)
     {
         static::$configs[$name] = $databaseConfig;
     }
@@ -309,13 +400,42 @@ public static function setConfig($databaseConfig, $name = 'default')
      * @param string $name
      * @return mixed
      */
-    public static function getConfig($name = 'default')
+    public static function getConfig($name = DB::CONN_PRIMARY)
     {
-        if (isset(static::$configs[$name])) {
+        if (static::hasConfig($name)) {
             return static::$configs[$name];
         }
     }
 
+    /**
+     * Check if a named connection config exists
+     */
+    public static function hasConfig($name = DB::CONN_PRIMARY): bool
+    {
+        return array_key_exists($name, static::$configs);
+    }
+
+    /**
+     * Get a replica database configuration key
+     * e.g. replica_01
+     */
+    public static function getReplicaConfigKey(int $replica): string
+    {
+        $len = strlen((string) DB::MAX_REPLICAS);
+        return 'replica_' . str_pad($replica, $len, '0', STR_PAD_LEFT);
+    }
+
+    /**
+     * Check if there are any replica configurations
+     */
+    public static function hasReplicaConfig(): bool
+    {
+        $configKeys = array_keys(static::$configs);
+        return !empty(array_filter($configKeys, function (string $key) {
+            return (bool) preg_match('#^replica_[0-9]+$#', $key);
+        }));
+    }
+
     /**
      * Returns true if a database connection has been attempted.
      * In particular, it lets the caller know if we're still so early in the execution pipeline that
@@ -335,8 +455,8 @@ public static function connection_attempted()
     public static function query($sql, $errorLevel = E_USER_ERROR)
     {
         DB::$lastQuery = $sql;
-
-        return DB::get_conn()->query($sql, $errorLevel);
+        $name = DB::getDynamicConnectionName($sql);
+        return DB::get_conn($name)->query($sql, $errorLevel);
     }
 
     /**
@@ -427,8 +547,8 @@ public static function inline_parameters($sql, $parameters)
     public static function prepared_query($sql, $parameters, $errorLevel = E_USER_ERROR)
     {
         DB::$lastQuery = $sql;
-
-        return DB::get_conn()->preparedQuery($sql, $parameters, $errorLevel);
+        $name = DB::getDynamicConnectionName($sql);
+        return DB::get_conn($name)->preparedQuery($sql, $parameters, $errorLevel);
     }
 
     /**
@@ -680,4 +800,63 @@ public static function alteration_message($message, $type = "")
     {
         DB::get_schema()->alterationMessage($message, $type);
     }
+
+    /**
+     * Get the name of the database connection to use for the given SQL query
+     * The 'dynamic' connection can be either the primary or a replica connection if configured
+     */
+    private static function getDynamicConnectionName(string $sql = ''): string
+    {
+        if (DB::willUsePrimary()) {
+            return DB::CONN_PRIMARY;
+        }
+        if (DB::isMutableSql($sql)) {
+            DB::$mustUsePrimary = true;
+            return DB::CONN_PRIMARY;
+        }
+        if (DB::$replicaConfigKey) {
+            return DB::$replicaConfigKey;
+        }
+        $name = DB::getRandomReplicaConfigKey();
+        DB::$replicaConfigKey = $name;
+        return $name;
+    }
+
+    /**
+     * Check if the given SQL query is mutable
+     */
+    private static function isMutableSql(string $sql): bool
+    {
+        $dbClass = DB::getConfig(DB::CONN_PRIMARY)['type'];
+        // This must use getServiceSpec() and not Injector::get/create() followed by
+        // getConnector() as this can remove the dbConn from a different connection
+        // under edge case conditions
+        $dbSpec = Injector::inst()->getServiceSpec($dbClass);
+        $connectorService = $dbSpec['properties']['connector'];
+        $connector = Injector::inst()->convertServiceProperty($connectorService);
+        return $connector->isQueryMutable($sql);
+    }
+
+    /**
+     * Get a random replica database configuration key from the available replica configurations
+     * The replica choosen will be used for the rest of the request, unless the primary connection
+     * is forced
+     */
+    private static function getRandomReplicaConfigKey(): string
+    {
+        $replicaNumbers = [];
+        for ($i = 1; $i <= DB::MAX_REPLICAS; $i++) {
+            $replicaKey = DB::getReplicaConfigKey($i);
+            if (!DB::hasConfig($replicaKey)) {
+                break;
+            }
+            $replicaNumbers[] = $i;
+        }
+        if (count($replicaNumbers) === 0) {
+            throw new RunTimeException('No replica configurations found');
+        }
+        // Choose a random replica
+        $index = rand(0, count($replicaNumbers) - 1);
+        return DB::getReplicaConfigKey($replicaNumbers[$index]);
+    }
 }
diff --git a/src/ORM/DataList.php b/src/ORM/DataList.php
index d703d3b9050..a65f5b161ef 100644
--- a/src/ORM/DataList.php
+++ b/src/ORM/DataList.php
@@ -237,7 +237,7 @@ public function setDataQueryParam($keyOrArray, $val = null)
      */
     public function sql(&$parameters = [])
     {
-        return $this->dataQuery->query()->sql($parameters);
+        return $this->dataQuery->sql($parameters);
     }
 
     /**
@@ -1040,7 +1040,7 @@ private function getEagerLoadVariables(string $relationChain, string $relationNa
 
     private function executeQuery(): Query
     {
-        $query = $this->dataQuery->query()->execute();
+        $query = $this->dataQuery->execute();
         $this->fetchEagerLoadRelations($query);
         return $query;
     }
diff --git a/src/ORM/DataObject.php b/src/ORM/DataObject.php
index 2b6bed1dae7..e0d2b755d61 100644
--- a/src/ORM/DataObject.php
+++ b/src/ORM/DataObject.php
@@ -145,6 +145,14 @@ class DataObject extends ModelData implements DataObjectInterface, i18nEntityPro
      */
     private static $default_classname = null;
 
+    /**
+     * Whether this DataObject class must only use the primary database and not a read-only replica
+     * Note that this will be only be enforced when using DataQuery::execute() or
+     * another method that uses calls DataQuery::execute() internally e.g. DataObject::get()
+     * This will not be enforced when using low-level ORM functionality to query data e.g. SQLSelect or DB::query()
+     */
+    private static bool $must_use_primary_db = false;
+
     /**
      * Data stored in this objects database record. An array indexed by fieldname.
      *
diff --git a/src/ORM/DataQuery.php b/src/ORM/DataQuery.php
index c9230986947..29f7153421b 100644
--- a/src/ORM/DataQuery.php
+++ b/src/ORM/DataQuery.php
@@ -10,6 +10,7 @@
 use SilverStripe\ORM\Queries\SQLConditionGroup;
 use SilverStripe\ORM\Queries\SQLSelect;
 use InvalidArgumentException;
+use SilverStripe\Core\Config\Config;
 
 /**
  * An object representing a query of data from the DataObject's supporting database.
@@ -449,7 +450,9 @@ protected function ensureSelectContainsOrderbyColumns($query, $originalSelect =
      */
     public function execute()
     {
-        return $this->getFinalisedQuery()->execute();
+        return $this->withCorrectDatabase(
+            fn() => $this->getFinalisedQuery()->execute()
+        );
     }
 
     /**
@@ -472,7 +475,9 @@ public function sql(&$parameters = [])
     public function count()
     {
         $quotedColumn = DataObject::getSchema()->sqlColumnForField($this->dataClass(), 'ID');
-        return $this->getFinalisedQuery()->count("DISTINCT {$quotedColumn}");
+        return $this->withCorrectDatabase(
+            fn() => $this->getFinalisedQuery()->count("DISTINCT {$quotedColumn}")
+        );
     }
 
     /**
@@ -501,7 +506,9 @@ public function exists(): bool
 
         // Wrap the whole thing in an "EXISTS"
         $sql = 'SELECT CASE WHEN EXISTS(' . $statement->sql($params) . ') THEN 1 ELSE 0 END';
-        $result = DB::prepared_query($sql, $params);
+        $result = $this->withCorrectDatabase(
+            fn() => DB::prepared_query($sql, $params)
+        );
         $row = $result->record();
         $result = reset($row);
 
@@ -582,7 +589,9 @@ public function sum($field)
      */
     public function aggregate($expression)
     {
-        return $this->getFinalisedQuery()->aggregate($expression)->execute()->value();
+        return $this->withCorrectDatabase(
+            fn() => $this->getFinalisedQuery()->aggregate($expression)->execute()->value()
+        );
     }
 
     /**
@@ -593,7 +602,9 @@ public function aggregate($expression)
      */
     public function firstRow()
     {
-        return $this->getFinalisedQuery()->firstRow();
+        return $this->withCorrectDatabase(
+            fn() => $this->getFinalisedQuery()->firstRow()
+        );
     }
 
     /**
@@ -604,7 +615,9 @@ public function firstRow()
      */
     public function lastRow()
     {
-        return $this->getFinalisedQuery()->lastRow();
+        return $this->withCorrectDatabase(
+            fn() => $this->getFinalisedQuery()->lastRow()
+        );
     }
 
     /**
@@ -1344,7 +1357,9 @@ public function column($field = 'ID')
         $query->selectField($fieldExpression, $field);
         $this->ensureSelectContainsOrderbyColumns($query, $originalSelect);
 
-        return $query->execute()->column($field);
+        return $this->withCorrectDatabase(
+            fn() => $query->execute()->column($field)
+        );
     }
 
     /**
@@ -1495,4 +1510,16 @@ private function getJoinTableName($class, $table)
 
         return $updated;
     }
+
+    /**
+     * Calls a callback on either the primary database or a replica, with respect to the configured
+     * value of `must_use_primary_db` on the current dataClass
+     */
+    private function withCorrectDatabase(callable $callback): mixed
+    {
+        if (Config::inst()->get($this->dataClass(), 'must_use_primary_db')) {
+            return DB::withPrimary($callback);
+        }
+        return $callback();
+    }
 }
diff --git a/src/Security/Group.php b/src/Security/Group.php
index 04466000957..6c10d32c073 100755
--- a/src/Security/Group.php
+++ b/src/Security/Group.php
@@ -88,6 +88,8 @@ class Group extends DataObject
 
     private static $table_name = "Group";
 
+    private static bool $must_use_primary_db = true;
+
     private static $indexes = [
         'Title' => true,
         'Code' => true,
diff --git a/src/Security/LoginAttempt.php b/src/Security/LoginAttempt.php
index 9ff1b31ce4e..b5142073c66 100644
--- a/src/Security/LoginAttempt.php
+++ b/src/Security/LoginAttempt.php
@@ -50,6 +50,8 @@ class LoginAttempt extends DataObject
 
     private static $table_name = "LoginAttempt";
 
+    private static bool $must_use_primary_db = true;
+
     /**
      * @param bool $includerelations Indicate if the labels returned include relation fields
      * @return array
diff --git a/src/Security/Member.php b/src/Security/Member.php
index e5d198650bb..94ea15ccfda 100644
--- a/src/Security/Member.php
+++ b/src/Security/Member.php
@@ -105,6 +105,8 @@ class Member extends DataObject
 
     private static $table_name = "Member";
 
+    private static bool $must_use_primary_db = true;
+
     private static $default_sort = '"Surname", "FirstName"';
 
     private static $indexes = [
diff --git a/src/Security/MemberPassword.php b/src/Security/MemberPassword.php
index 8488671f6c6..cebb2cce116 100644
--- a/src/Security/MemberPassword.php
+++ b/src/Security/MemberPassword.php
@@ -27,6 +27,8 @@ class MemberPassword extends DataObject
 
     private static $table_name = "MemberPassword";
 
+    private static bool $must_use_primary_db = true;
+
     /**
      * Log a password change from the given member.
      * Call MemberPassword::log($this) from within Member whenever the password is changed.
diff --git a/src/Security/Permission.php b/src/Security/Permission.php
index 0fec4b75207..a249b3dadfc 100644
--- a/src/Security/Permission.php
+++ b/src/Security/Permission.php
@@ -46,6 +46,8 @@ class Permission extends DataObject implements TemplateGlobalProvider, Resettabl
 
     private static $table_name = "Permission";
 
+    private static bool $must_use_primary_db = true;
+
     /**
      * This is the value to use for the "Type" field if a permission should be
      * granted.
@@ -233,15 +235,15 @@ public static function checkMember($member, $code, $arg = "any", $strict = true)
         }
 
         // Raw SQL for efficiency
-        $permission = DB::prepared_query(
+        $permission = DB::withPrimary(fn() => DB::prepared_query(
             "SELECT \"ID\"
-			FROM \"Permission\"
-			WHERE (
-				\"Code\" IN ($codeClause $adminClause)
-				AND \"Type\" = ?
-				AND \"GroupID\" IN ($groupClause)
-				$argClause
-			)",
+            FROM \"Permission\"
+            WHERE (
+                \"Code\" IN ($codeClause $adminClause)
+                AND \"Type\" = ?
+                AND \"GroupID\" IN ($groupClause)
+                $argClause
+            )",
             array_merge(
                 $codeParams,
                 $adminParams,
@@ -249,7 +251,7 @@ public static function checkMember($member, $code, $arg = "any", $strict = true)
                 $groupParams,
                 $argParams
             )
-        )->value();
+        )->value());
 
         if ($permission) {
             return $permission;
@@ -257,15 +259,15 @@ public static function checkMember($member, $code, $arg = "any", $strict = true)
 
         // Strict checking disabled?
         if (!static::config()->strict_checking || !$strict) {
-            $hasPermission = DB::prepared_query(
+            $hasPermission = DB::withPrimary(fn() => DB::prepared_query(
                 "SELECT COUNT(*)
-				FROM \"Permission\"
-				WHERE (
-					\"Code\" IN ($codeClause) AND
-					\"Type\" = ?
-				)",
+                FROM \"Permission\"
+                WHERE (
+                    \"Code\" IN ($codeClause) AND
+                    \"Type\" = ?
+                )",
                 array_merge($codeParams, [Permission::GRANT_PERMISSION])
-            )->value();
+            )->value());
 
             if (!$hasPermission) {
                 return false;
@@ -288,25 +290,29 @@ public static function permissions_for_member($memberID)
         if ($groupList) {
             $groupCSV = implode(", ", $groupList);
 
-            $allowed = array_unique(DB::query("
-				SELECT \"Code\"
-				FROM \"Permission\"
-				WHERE \"Type\" = " . Permission::GRANT_PERMISSION . " AND \"GroupID\" IN ($groupCSV)
-
-				UNION
-
-				SELECT \"Code\"
-				FROM \"PermissionRoleCode\" PRC
-				INNER JOIN \"PermissionRole\" PR ON PRC.\"RoleID\" = PR.\"ID\"
-				INNER JOIN \"Group_Roles\" GR ON GR.\"PermissionRoleID\" = PR.\"ID\"
-				WHERE \"GroupID\" IN ($groupCSV)
-			")->column() ?? []);
-
-            $denied = array_unique(DB::query("
-				SELECT \"Code\"
-				FROM \"Permission\"
-				WHERE \"Type\" = " . Permission::DENY_PERMISSION . " AND \"GroupID\" IN ($groupCSV)
-			")->column() ?? []);
+            $allowed = array_unique(
+                DB::withPrimary(fn() => DB::query("
+                    SELECT \"Code\"
+                    FROM \"Permission\"
+                    WHERE \"Type\" = " . Permission::GRANT_PERMISSION . " AND \"GroupID\" IN ($groupCSV)
+
+                    UNION
+
+                    SELECT \"Code\"
+                    FROM \"PermissionRoleCode\" PRC
+                    INNER JOIN \"PermissionRole\" PR ON PRC.\"RoleID\" = PR.\"ID\"
+                    INNER JOIN \"Group_Roles\" GR ON GR.\"PermissionRoleID\" = PR.\"ID\"
+                    WHERE \"GroupID\" IN ($groupCSV)
+                "))->column() ?? []
+            );
+
+            $denied = array_unique(
+                DB::withPrimary(fn() => DB::query("
+                    SELECT \"Code\"
+                    FROM \"Permission\"
+                    WHERE \"Type\" = " . Permission::DENY_PERMISSION . " AND \"GroupID\" IN ($groupCSV)
+                "))->column() ?? []
+            );
 
             return array_diff($allowed ?? [], $denied);
         }
@@ -584,7 +590,9 @@ public static function get_codes($grouped = true)
                 $flatCodeArray[] = $code;
             }
         }
-        $otherPerms = DB::query("SELECT DISTINCT \"Code\" From \"Permission\" WHERE \"Code\" != ''")->column();
+        $otherPerms = DB::withPrimary(
+            fn() => DB::query("SELECT DISTINCT \"Code\" From \"Permission\" WHERE \"Code\" != ''")->column()
+        );
 
         if ($otherPerms) {
             foreach ($otherPerms as $otherPerm) {
diff --git a/src/Security/PermissionRole.php b/src/Security/PermissionRole.php
index 92d99b38ef4..ccfc02fcaf7 100644
--- a/src/Security/PermissionRole.php
+++ b/src/Security/PermissionRole.php
@@ -40,6 +40,8 @@ class PermissionRole extends DataObject
 
     private static $table_name = "PermissionRole";
 
+    private static bool $must_use_primary_db = true;
+
     private static $default_sort = '"Title"';
 
     private static $singular_name = 'Role';
diff --git a/src/Security/PermissionRoleCode.php b/src/Security/PermissionRoleCode.php
index 8d7e2979bd0..8893382419e 100644
--- a/src/Security/PermissionRoleCode.php
+++ b/src/Security/PermissionRoleCode.php
@@ -23,6 +23,8 @@ class PermissionRoleCode extends DataObject
     ];
 
     private static $table_name = "PermissionRoleCode";
+
+    private static bool $must_use_primary_db = true;
     
     private static $indexes = [
         "Code" => true,
diff --git a/src/Security/RememberLoginHash.php b/src/Security/RememberLoginHash.php
index 8af22057353..1895445b226 100644
--- a/src/Security/RememberLoginHash.php
+++ b/src/Security/RememberLoginHash.php
@@ -44,6 +44,8 @@ class RememberLoginHash extends DataObject
 
     private static $table_name = "RememberLoginHash";
 
+    private static bool $must_use_primary_db = true;
+
     /**
      * Determines if logging out on one device also clears existing login tokens
      * on all other devices owned by the member.
diff --git a/src/Security/Security.php b/src/Security/Security.php
index 231b6d7d5e8..b3539398dfb 100644
--- a/src/Security/Security.php
+++ b/src/Security/Security.php
@@ -437,6 +437,12 @@ public static function permissionFailure($controller = null, $messageSet = null)
      */
     public static function setCurrentUser($currentUser = null)
     {
+        // Always use the primary database and not a replica if a CMS user is logged in
+        // This is to ensure that when viewing content on the frontend it is always
+        // up to date i.e. not from an unsynced replica
+        if ($currentUser && Permission::checkMember($currentUser, 'CMS_ACCESS')) {
+            DB::setMustUsePrimary();
+        }
         Security::$currentUser = $currentUser;
     }
 
diff --git a/tests/php/Core/KernelTest.php b/tests/php/Core/KernelTest.php
index 42f05c2714e..c779f349521 100644
--- a/tests/php/Core/KernelTest.php
+++ b/tests/php/Core/KernelTest.php
@@ -11,6 +11,10 @@
 use SilverStripe\Core\Injector\InjectorLoader;
 use SilverStripe\Core\Kernel;
 use SilverStripe\Dev\SapphireTest;
+use SilverStripe\Core\Environment;
+use ReflectionClass;
+use SilverStripe\ORM\DB;
+use ReflectionObject;
 
 class KernelTest extends SapphireTest
 {
@@ -81,4 +85,32 @@ public function testInvalidConfigDetection()
 
         $kernel->getConfigLoader()->getManifest();
     }
+    
+    public function testReplicaDatabaseVarsLoaded()
+    {
+        // Set environment variables for a fake replica database
+        Environment::setEnv('SS_DATABASE_SERVER_REPLICA_01', 'the-moon');
+        Environment::setEnv('SS_DATABASE_USERNAME_REPLICA_01', 'alien');
+        Environment::setEnv('SS_DATABASE_PASSWORD_REPLICA_01', 'hi_people');
+        // Get the CoreKernel
+        /** @var Kernel $kernel */
+        $kernel = Injector::inst()->get(Kernel::class);
+        /** @var CoreKernel $coreKernel */
+        $coreKernel = $kernel->nest();
+        $this->assertTrue(is_a($coreKernel, CoreKernel::class));
+        // Boot the database environment variables
+        $reflector = new ReflectionObject($coreKernel);
+        $method = $reflector->getMethod('bootDatabaseEnvVars');
+        $method->setAccessible(true);
+        $method->invoke($coreKernel);
+        // Assert DB config was updated
+        $default = DB::getConfig(DB::CONN_PRIMARY);
+        $configs = (new ReflectionClass(DB::class))->getStaticPropertyValue('configs');
+        $this->assertSame([
+            'type' => $default['type'],
+            'server' => 'the-moon',
+            'username' => 'alien',
+            'password' => 'hi_people',
+        ], $configs['replica_01']);
+    }
 }
diff --git a/tests/php/ORM/DBReplicaTest.php b/tests/php/ORM/DBReplicaTest.php
new file mode 100644
index 00000000000..d601e8b6540
--- /dev/null
+++ b/tests/php/ORM/DBReplicaTest.php
@@ -0,0 +1,260 @@
+<?php
+
+namespace SilverStripe\ORM\Tests;
+
+use ReflectionClass;
+use SilverStripe\Dev\FunctionalTest;
+use SilverStripe\ORM\DB;
+use SilverStripe\Control\Director;
+use SilverStripe\Security\Security;
+use SilverStripe\Core\Config\Config;
+use SilverStripe\ORM\Tests\DBReplicaTest\TestController;
+use SilverStripe\ORM\Tests\DBReplicaTest\TestObject;
+use SilverStripe\Security\Group;
+use SilverStripe\Security\Member;
+use SilverStripe\ORM\DataQuery;
+use PHPUnit\Framework\Attributes\DataProvider;
+
+class DBReplicaTest extends FunctionalTest
+{
+    protected static $extra_dataobjects = [
+        TestObject::class,
+    ];
+
+    protected static $fixture_file = 'DBReplicaTest.yml';
+
+    protected function setUp(): void
+    {
+        parent::setUp();
+        $this->setupConfigsAndConnections(true);
+        // Set DB:$mustUsePrimary to true to allow using replicas
+        // This is disabled by default in SapphireTest::setUpBeforeClass()
+        // Also reset mustUsePrimary after using mutable sql to create yml fixtures
+        // and also because by default an ADMIN user is logged in when using fixtures in SapphireTest::setUp()
+        // and also prevent tests from affecting subsequent tests
+        (new ReflectionClass(DB::class))->setStaticPropertyValue('mustUsePrimary', false);
+    }
+
+    protected function tearDown(): void
+    {
+        $this->setupConfigsAndConnections(false);
+        // Reset DB:$mustUsePrimary to true which is the default set by SapphireTest::setUpBeforeClass()
+        (new ReflectionClass(DB::class))->setStaticPropertyValue('mustUsePrimary', true);
+        parent::tearDown();
+    }
+    
+    public function testUsesReplica(): void
+    {
+        // Assert uses replica by default
+        TestObject::get()->count();
+        $this->assertSame('replica_01', $this->getLastConnectionName());
+        // Assert uses primary when using withPrimary()
+        DB::withPrimary(fn() => TestObject::get()->count());
+        $this->assertSame(DB::CONN_PRIMARY, $this->getLastConnectionName());
+        // Assert that withPrimary() was only temporary
+        TestObject::get()->count();
+        $this->assertSame('replica_01', $this->getLastConnectionName());
+        // Assert DB::setMustUsePrimary() forces primary from now on
+        DB::setMustUsePrimary();
+        TestObject::get()->count();
+        $this->assertSame(DB::CONN_PRIMARY, $this->getLastConnectionName());
+    }
+
+    public function testMutableSql(): void
+    {
+        // Assert that using mutable sql in an ORM method with a dataclass uses primary
+        TestObject::create(['Title' => 'testing'])->write();
+        $this->assertSame(DB::CONN_PRIMARY, $this->getLastConnectionName());
+        // Assert that now all subsequent queries use primary
+        TestObject::get()->count();
+        $this->assertSame(DB::CONN_PRIMARY, $this->getLastConnectionName());
+    }
+
+    public function testMutableSqlDbQuery(): void
+    {
+        // Assert that using mutable sql in DB::query() uses primary
+        DB::query('INSERT INTO "DBReplicaTest_TestObject" ("Title") VALUES (\'testing\')');
+        $this->assertSame(DB::CONN_PRIMARY, $this->getLastConnectionName());
+        // Assert that now all subsequent queries use primary
+        TestObject::get()->count();
+        $this->assertSame(DB::CONN_PRIMARY, $this->getLastConnectionName());
+    }
+
+    public function testMutableSqlDbPreparedQuery(): void
+    {
+        // Assert that using mutable sql in DB::prepared_query() uses primary
+        DB::prepared_query('INSERT INTO "DBReplicaTest_TestObject" ("Title") VALUES (?)', ['testing']);
+        $this->assertSame(DB::CONN_PRIMARY, $this->getLastConnectionName());
+        // Assert that now all subsequent queries use primary
+        TestObject::get()->count();
+        $this->assertSame(DB::CONN_PRIMARY, $this->getLastConnectionName());
+    }
+
+    #[DataProvider('provideSetCurrentUser')]
+    public function testSetCurrentUser(string $firstName, string $expected): void
+    {
+        $member = Member::get()->find('FirstName', $firstName);
+        Security::setCurrentUser($member);
+        TestObject::get()->count();
+        $this->assertSame($expected, $this->getLastConnectionName());
+    }
+
+    public function testDataObjectMustUsePrimaryDb(): void
+    {
+        // Assert that DataList::getIterator() respect DataObject.must_use_primary_db
+        foreach (TestObject::get() as $object) {
+            $object->Title = 'test2';
+        }
+        $this->assertSame('replica_01', $this->getLastConnectionName());
+        foreach (Group::get() as $group) {
+            $group->Title = 'test2';
+        }
+        $this->assertSame(DB::CONN_PRIMARY, $this->getLastConnectionName());
+        // Assert that DataQuery methods without params respect DataObject.must_use_primary_db
+        $methods = [
+            'count',
+            'exists',
+            'firstRow',
+            'lastRow'
+        ];
+        foreach ($methods as $method) {
+            (new DataQuery(TestObject::class))->$method();
+            $this->assertSame('replica_01', $this->getLastConnectionName(), "method is $method");
+            (new DataQuery(Group::class))->$method();
+            $this->assertSame(DB::CONN_PRIMARY, $this->getLastConnectionName(), "method is $method");
+        }
+        // Assert that DataQuery methods with a param respect DataObject.must_use_primary_db
+        $methods = [
+            'max',
+            'min',
+            'avg',
+            'sum',
+            'column',
+        ];
+        foreach ($methods as $method) {
+            (new DataQuery(TestObject::class))->$method('ID');
+            $this->assertSame('replica_01', $this->getLastConnectionName(), "method is $method");
+            (new DataQuery(Group::class))->$method('ID');
+            $this->assertSame(DB::CONN_PRIMARY, $this->getLastConnectionName(), "method is $method");
+        }
+    }
+
+    public static function provideSetCurrentUser(): array
+    {
+        return [
+            'non_cms_user' => [
+                'firstName' => 'random',
+                'expected' => 'replica_01'
+            ],
+            'cms_user' => [
+                'firstName' => 'cmsuser',
+                'expected' => DB::CONN_PRIMARY
+            ],
+        ];
+    }
+
+    public static function provideRoutes(): array
+    {
+        return [
+            'normal_route' => [
+                'path' => 'test',
+                'expected' => 'replica_01'
+            ],
+            'security_route' => [
+                'path' => 'Security/login',
+                'expected' => DB::CONN_PRIMARY
+            ],
+            'dev_route' => [
+                'path' => 'dev/tasks',
+                'expected' => DB::CONN_PRIMARY
+            ],
+            'dev_in_path_but_not_dev_route' => [
+                'path' => 'test/dev',
+                'expected' => 'replica_01'
+            ],
+        ];
+    }
+
+    #[DataProvider('provideRoutes')]
+    public function testRoutes(string $path, string $expected): void
+    {
+        // Create a custom rule to test our controller that should default to using a replica
+        $rules = Config::inst()->get(Director::class, 'rules');
+        $rules['test'] = TestController::class;
+        // Ensure that routes staring with '$' are at the bottom of the assoc array index and don't override
+        // our new 'test' route
+        uksort($rules, fn($a, $b) => str_starts_with($a, '$') ? 1 : (str_starts_with($b, '$') ? -1 : 0));
+        $this->get($path);
+        $this->assertSame($expected, $this->getLastConnectionName());
+    }
+
+    public static function provideHasReplicaConfig(): array
+    {
+        return [
+            'no_replica' => [
+                'includeReplica' => false,
+                'expected' => false
+            ],
+            'with_replica' => [
+                'includeReplica' => true,
+                'expected' => true
+            ],
+        ];
+    }
+
+    #[DataProvider('provideHasReplicaConfig')]
+    public function testHasReplicaConfig(bool $includeReplica, bool $expected): void
+    {
+        $this->assertTrue(DB::hasReplicaConfig());
+        $primaryConfig = DB::getConfig(DB::CONN_PRIMARY);
+        $config = [DB::CONN_PRIMARY => $primaryConfig];
+        if ($includeReplica) {
+            $config['replica_01'] = $primaryConfig;
+        }
+        (new ReflectionClass(DB::class))->setStaticPropertyValue('configs', $config);
+        $this->assertSame($expected, DB::hasReplicaConfig());
+    }
+
+    public function testHasConfig(): void
+    {
+        $this->assertFalse(DB::hasConfig('lorem'));
+        DB::setConfig(['type' => 'lorem'], 'lorem');
+        $this->assertTrue(DB::hasConfig('lorem'));
+    }
+
+    public function testGetReplicaConfigKey(): void
+    {
+        $this->assertSame('replica_03', DB::getReplicaConfigKey(3));
+        $this->assertSame('replica_58', DB::getReplicaConfigKey(58));
+    }
+
+    /**
+     * Using reflection, set DB::configs and DB::connections with a fake a replica connection
+     * that points to the same connection as the primary connection.
+     */
+    private function setupConfigsAndConnections($includeReplica = true): void
+    {
+        $reflector = new ReflectionClass(DB::class);
+        $primaryConfig = DB::getConfig(DB::CONN_PRIMARY);
+        $configs = [DB::CONN_PRIMARY => $primaryConfig];
+        if ($includeReplica) {
+            $configs['replica_01'] = $primaryConfig;
+        }
+        $reflector->setStaticPropertyValue('configs', $configs);
+        // Create connections
+        $primaryConnection = DB::get_conn(DB::CONN_PRIMARY);
+        $connections = [DB::CONN_PRIMARY => $primaryConnection];
+        if ($includeReplica) {
+            $connections['replica_01'] = $primaryConnection;
+        }
+        $reflector->setStaticPropertyValue('connections', $connections);
+    }
+
+    /**
+     * Get the last connection name used by the DB class. This shows if a replica was used.
+     */
+    private function getLastConnectionName(): string
+    {
+        return (new ReflectionClass(DB::class))->getStaticPropertyValue('lastConnectionName');
+    }
+}
diff --git a/tests/php/ORM/DBReplicaTest.yml b/tests/php/ORM/DBReplicaTest.yml
new file mode 100644
index 00000000000..4f7dd7f8c90
--- /dev/null
+++ b/tests/php/ORM/DBReplicaTest.yml
@@ -0,0 +1,18 @@
+SilverStripe\ORM\Tests\DBReplicaTest\TestObject:
+  test:
+    Title: 'test'
+SilverStripe\Security\Permission:
+  test:
+    Code: 'CMS_ACCESS_Something'
+SilverStripe\Security\Group:
+  test:
+    Title: 'test'
+    Permissions:
+      - =>SilverStripe\Security\Permission.test
+SilverStripe\Security\Member:
+  cmsuser:
+    FirstName: 'CmsUser'
+    Groups:
+      - =>SilverStripe\Security\Group.test
+  random:
+    FirstName: 'Random'
diff --git a/tests/php/ORM/DBReplicaTest/TestController.php b/tests/php/ORM/DBReplicaTest/TestController.php
new file mode 100644
index 00000000000..56cdc92eade
--- /dev/null
+++ b/tests/php/ORM/DBReplicaTest/TestController.php
@@ -0,0 +1,18 @@
+<?php
+
+namespace SilverStripe\ORM\Tests\DBReplicaTest;
+
+use SilverStripe\Control\Controller;
+use SilverStripe\Dev\TestOnly;
+
+class TestController extends Controller implements TestOnly
+{
+    public function index()
+    {
+        // Make a call to the database
+        TestObject::get()->count();
+        $response = $this->getResponse();
+        $response->setBody('DB_REPLICA_TEST_CONTROLLER');
+        return $response;
+    }
+}
diff --git a/tests/php/ORM/DBReplicaTest/TestObject.php b/tests/php/ORM/DBReplicaTest/TestObject.php
new file mode 100644
index 00000000000..aac3f295c74
--- /dev/null
+++ b/tests/php/ORM/DBReplicaTest/TestObject.php
@@ -0,0 +1,16 @@
+<?php
+
+namespace SilverStripe\ORM\Tests\DBReplicaTest;
+
+use SilverStripe\Control\Controller;
+use SilverStripe\Dev\TestOnly;
+use SilverStripe\ORM\DataObject;
+
+class TestObject extends DataObject implements TestOnly
+{
+    private static $db = [
+        'Title' => 'Varchar',
+    ];
+
+    private static $table_name = 'DBReplicaTest_TestObject';
+}
diff --git a/tests/php/ORM/MySQLiConnectorTest.php b/tests/php/ORM/MySQLiConnectorTest.php
index 3d7ac8d0a16..a93dc3e03a3 100644
--- a/tests/php/ORM/MySQLiConnectorTest.php
+++ b/tests/php/ORM/MySQLiConnectorTest.php
@@ -41,7 +41,7 @@ public function setUp(): void
     {
         parent::setUp();
 
-        $config = DB::getConfig();
+        $config = DB::getConfig(DB::CONN_PRIMARY);
 
         if (strtolower(substr($config['type'] ?? '', 0, 5)) !== 'mysql') {
             $this->markTestSkipped("The test only relevant for MySQL - but $config[type] is in use");