[stable28] feat(database): Primary/replica split

config/config.sample.php

@@ -151,6 +151,14 @@
  */
 'dbpersistent' => '',
 
+/**
+ * Specify read only replicas to be used by Nextcloud when querying the database
+ */
+'dbreplica' => [
+	['user' => 'nextcloud', 'password' => 'password1', 'host' => 'replica1', 'dbname' => ''],
+	['user' => 'nextcloud', 'password' => 'password2', 'host' => 'replica2', 'dbname' => ''],
+],
+
 /**
  * Indicates whether the Nextcloud instance was installed successfully; ``true``
  * indicates a successful installation, and ``false`` indicates an unsuccessful
@@ -986,6 +994,15 @@
  */
 'loglevel_frontend' => 2,
 
+/**
+ * Loglevel used by the dirty database query detection. Useful to identify
+ * potential database bugs in production. Set this to loglevel or higher to
+ * see dirty queries in the logs.
+ *
+ * Defaults to ``0`` (debug)
+ */
+'loglevel_dirty_database_queries' => 0,
+
 /**
  * If you maintain different instances and aggregate the logs, you may want
  * to distinguish between them. ``syslog_tag`` can be set per instance

lib/private/DB/Connection.php

@@ -38,6 +38,7 @@
 use Doctrine\Common\EventManager;
 use Doctrine\DBAL\Cache\QueryCacheProfile;
 use Doctrine\DBAL\Configuration;
+use Doctrine\DBAL\Connections\PrimaryReadReplicaConnection;
 use Doctrine\DBAL\Driver;
 use Doctrine\DBAL\Exception;
 use Doctrine\DBAL\Platforms\MySQLPlatform;
@@ -53,9 +54,11 @@
 use OCP\IRequestId;
 use OCP\PreConditionNotMetException;
 use OCP\Profiler\IProfiler;
+use Psr\Clock\ClockInterface;
 use Psr\Log\LoggerInterface;
+use function in_array;
 
-class Connection extends \Doctrine\DBAL\Connection {
+class Connection extends PrimaryReadReplicaConnection {
 	/** @var string */
 	protected $tablePrefix;
 
@@ -65,6 +68,8 @@ class Connection extends \Doctrine\DBAL\Connection {
 	/** @var SystemConfig */
 	private $systemConfig;
 
+	private ClockInterface $clock;
+
 	private LoggerInterface $logger;
 
 	protected $lockedTable = null;
@@ -78,6 +83,11 @@ class Connection extends \Doctrine\DBAL\Connection {
 	/** @var DbDataCollector|null */
 	protected $dbDataCollector = null;
 
+	protected ?float $transactionActiveSince = null;
+
+	/** @var array<string, int> */
+	protected $tableDirtyWrites = [];
+
 	/**
 	 * Initializes a new instance of the Connection class.
 	 *
@@ -103,6 +113,7 @@ public function __construct(
 		$this->tablePrefix = $params['tablePrefix'];
 
 		$this->systemConfig = \OC::$server->getSystemConfig();
+		$this->clock = \OCP\Server::get(ClockInterface::class);
 		$this->logger = \OC::$server->get(LoggerInterface::class);
 
 		/** @var \OCP\Profiler\IProfiler */
@@ -119,7 +130,7 @@ public function __construct(
 	/**
 	 * @throws Exception
 	 */
-	public function connect() {
+	public function connect($connectionName = null) {
 		try {
 			if ($this->_conn) {
 				/** @psalm-suppress InternalMethod */
@@ -254,13 +265,59 @@ public function prepare($sql, $limit = null, $offset = null): Statement {
 	 * @throws \Doctrine\DBAL\Exception
 	 */
 	public function executeQuery(string $sql, array $params = [], $types = [], QueryCacheProfile $qcp = null): Result {
+		$tables = $this->getQueriedTables($sql);
+		$now = $this->clock->now()->getTimestamp();
+		$dirtyTableWrites = [];
+		foreach ($tables as $table) {
+			$lastAccess = $this->tableDirtyWrites[$table] ?? 0;
+			// Only very recent writes are considered dirty
+			if ($lastAccess >= ($now - 3)) {
+				$dirtyTableWrites[] = $table;
+			}
+		}
+		if ($this->isTransactionActive()) {
+			// Transacted queries go to the primary. The consistency of the primary guarantees that we can not run
+			// into a dirty read.
+		} elseif (count($dirtyTableWrites) === 0) {
+			// No tables read that could have been written already in the same request and no transaction active
+			// so we can switch back to the replica for reading as long as no writes happen that switch back to the primary
+			// We cannot log here as this would log too early in the server boot process
+			$this->ensureConnectedToReplica();
+		} else {
+			// Read to a table that has been written to previously
+			// While this might not necessarily mean that we did a read after write it is an indication for a code path to check
+			$this->logger->log(
+				(int) ($this->systemConfig->getValue('loglevel_dirty_database_queries', null) ?? 0),
+				'dirty table reads: ' . $sql,
+				[
+					'tables' => array_keys($this->tableDirtyWrites),
+					'reads' => $tables,
+					'exception' => new \Exception(),
+				],
+			);
+			// To prevent a dirty read on a replica that is slightly out of sync, we
+			// switch back to the primary. This is detrimental for performance but
+			// safer for consistency.
+			$this->ensureConnectedToPrimary();
+		}
+
 		$sql = $this->replaceTablePrefix($sql);
 		$sql = $this->adapter->fixupStatement($sql);
 		$this->queriesExecuted++;
 		$this->logQueryToFile($sql);
 		return parent::executeQuery($sql, $params, $types, $qcp);
 	}
 
+	/**
+	 * Helper function to get the list of tables affected by a given query
+	 * used to track dirty tables that received a write with the current request
+	 */
+	private function getQueriedTables(string $sql): array {
+		$re = '/(\*PREFIX\*\w+)/mi';
+		preg_match_all($re, $sql, $matches);
+		return array_map([$this, 'replaceTablePrefix'], $matches[0] ?? []);
+	}
+
 	/**
 	 * @throws Exception
 	 */
@@ -287,6 +344,10 @@ public function executeUpdate(string $sql, array $params = [], array $types = []
 	 * @throws \Doctrine\DBAL\Exception
 	 */
 	public function executeStatement($sql, array $params = [], array $types = []): int {
+		$tables = $this->getQueriedTables($sql);
+		foreach ($tables as $table) {
+			$this->tableDirtyWrites[$table] = $this->clock->now()->getTimestamp();
+		}
 		$sql = $this->replaceTablePrefix($sql);
 		$sql = $this->adapter->fixupStatement($sql);
 		$this->queriesExecuted++;
@@ -302,6 +363,11 @@ protected function logQueryToFile(string $sql): void {
 				$prefix .= \OC::$server->get(IRequestId::class)->getId() . "\t";
 			}
 
+			// FIXME:  Improve to log the actual target db host
+			$isPrimary = $this->connections['primary'] === $this->_conn;
+			$prefix .= ' ' . ($isPrimary === true ? 'primary' : 'replica') . ' ';
+			$prefix .= ' ' . $this->getTransactionNestingLevel() . ' ';
+
 			file_put_contents(
 				$this->systemConfig->getValue('query_log_file', ''),
 				$prefix . $sql . "\n",
@@ -603,4 +669,35 @@ private function getMigrator() {
 			return new Migrator($this, $config, $dispatcher);
 		}
 	}
+
+	public function beginTransaction() {
+		if (!$this->inTransaction()) {
+			$this->transactionActiveSince = microtime(true);
+		}
+		return parent::beginTransaction();
+	}
+
+	public function commit() {
+		$result = parent::commit();
+		if ($this->getTransactionNestingLevel() === 0) {
+			$timeTook = microtime(true) - $this->transactionActiveSince;
+			$this->transactionActiveSince = null;
+			if ($timeTook > 1) {
+				$this->logger->warning('Transaction took ' . $timeTook . 's', ['exception' => new \Exception('Transaction took ' . $timeTook . 's')]);
+			}
+		}
+		return $result;
+	}
+
+	public function rollBack() {
+		$result = parent::rollBack();
+		if ($this->getTransactionNestingLevel() === 0) {
+			$timeTook = microtime(true) - $this->transactionActiveSince;
+			$this->transactionActiveSince = null;
+			if ($timeTook > 1) {
+				$this->logger->warning('Transaction rollback took longer than 1s: ' . $timeTook, ['exception' => new \Exception('Long running transaction rollback')]);
+			}
+		}
+		return $result;
+	}
 }

lib/private/DB/ConnectionFactory.php

@@ -32,7 +32,6 @@
 use Doctrine\DBAL\Configuration;
 use Doctrine\DBAL\DriverManager;
 use Doctrine\DBAL\Event\Listeners\OracleSessionInit;
-use Doctrine\DBAL\Event\Listeners\SQLSessionInit;
 use OC\SystemConfig;
 
 /**
@@ -127,11 +126,8 @@ public function getConnection($type, $additionalConnectionParams) {
 		$normalizedType = $this->normalizeType($type);
 		$eventManager = new EventManager();
 		$eventManager->addEventSubscriber(new SetTransactionIsolationLevel());
+		$additionalConnectionParams = array_merge($this->createConnectionParams(), $additionalConnectionParams);
 		switch ($normalizedType) {
-			case 'mysql':
-				$eventManager->addEventSubscriber(
-					new SQLSessionInit("SET SESSION AUTOCOMMIT=1"));
-				break;
 			case 'oci':
 				$eventManager->addEventSubscriber(new OracleSessionInit);
 				// the driverOptions are unused in dbal and need to be mapped to the parameters
@@ -159,7 +155,7 @@ public function getConnection($type, $additionalConnectionParams) {
 		}
 		/** @var Connection $connection */
 		$connection = DriverManager::getConnection(
-			array_merge($this->getDefaultConnectionParams($type), $additionalConnectionParams),
+			$additionalConnectionParams,
 			new Configuration(),
 			$eventManager
 		);
@@ -195,10 +191,10 @@ public function isValidType($type) {
 	public function createConnectionParams(string $configPrefix = '') {
 		$type = $this->config->getValue('dbtype', 'sqlite');
 
-		$connectionParams = [
+		$connectionParams = array_merge($this->getDefaultConnectionParams($type), [
 			'user' => $this->config->getValue($configPrefix . 'dbuser', $this->config->getValue('dbuser', '')),
 			'password' => $this->config->getValue($configPrefix . 'dbpassword', $this->config->getValue('dbpassword', '')),
-		];
+		]);
 		$name = $this->config->getValue($configPrefix . 'dbname', $this->config->getValue('dbname', self::DEFAULT_DBNAME));
 
 		if ($this->normalizeType($type) === 'sqlite3') {
@@ -237,7 +233,11 @@ public function createConnectionParams(string $configPrefix = '') {
 			$connectionParams['persistent'] = true;
 		}
 
-		return $connectionParams;
+		$replica = $this->config->getValue('dbreplica', []) ?: [$connectionParams];
+		return array_merge($connectionParams, [
+			'primary' => $connectionParams,
+			'replica' => $replica,
+		]);
 	}
 
 	/**

lib/private/DB/SetTransactionIsolationLevel.php

@@ -26,8 +26,10 @@
 namespace OC\DB;
 
 use Doctrine\Common\EventSubscriber;
+use Doctrine\DBAL\Connections\PrimaryReadReplicaConnection;
 use Doctrine\DBAL\Event\ConnectionEventArgs;
 use Doctrine\DBAL\Events;
+use Doctrine\DBAL\Platforms\MySQLPlatform;
 use Doctrine\DBAL\TransactionIsolationLevel;
 
 class SetTransactionIsolationLevel implements EventSubscriber {
@@ -36,7 +38,13 @@ class SetTransactionIsolationLevel implements EventSubscriber {
 	 * @return void
 	 */
 	public function postConnect(ConnectionEventArgs $args) {
-		$args->getConnection()->setTransactionIsolation(TransactionIsolationLevel::READ_COMMITTED);
+		$connection = $args->getConnection();
+		if ($connection instanceof PrimaryReadReplicaConnection && $connection->isConnectedToPrimary()) {
+			$connection->setTransactionIsolation(TransactionIsolationLevel::READ_COMMITTED);
+			if ($connection->getDatabasePlatform() instanceof MySQLPlatform) {
+				$connection->executeStatement('SET SESSION AUTOCOMMIT=1');
+			}
+		}
 	}
 
 	public function getSubscribedEvents() {

lib/private/Server.php

@@ -843,8 +843,7 @@ public function __construct($webRoot, \OC\Config $config) {
 			if (!$factory->isValidType($type)) {
 				throw new \OC\DatabaseException('Invalid database type');
 			}
-			$connectionParams = $factory->createConnectionParams();
-			$connection = $factory->getConnection($type, $connectionParams);
+			$connection = $factory->getConnection($type, []);
 			return $connection;
 		});
 		/** @deprecated 19.0.0 */

lib/private/Setup/AbstractDatabase.php

@@ -140,10 +140,12 @@ protected function connect(array $configOverwrite = []): Connection {
 			}
 			$connectionParams['host'] = $host;
 		}
-
 		$connectionParams = array_merge($connectionParams, $configOverwrite);
+		$connectionParams = array_merge($connectionParams, ['primary' => $connectionParams, 'replica' => [$connectionParams]]);
 		$cf = new ConnectionFactory($this->config);
-		return $cf->getConnection($this->config->getValue('dbtype', 'sqlite'), $connectionParams);
+		$connection = $cf->getConnection($this->config->getValue('dbtype', 'sqlite'), $connectionParams);
+		$connection->ensureConnectedToPrimary();
+		return $connection;
 	}
 
 	/**