fix(propagator): Improve lock behavior of propagator

lib/composer/composer/LICENSE

@@ -1,3 +1,4 @@
+
 Copyright (c) Nils Adermann, Jordi Boggiano
 
 Permission is hereby granted, free of charge, to any person obtaining a copy
@@ -17,3 +18,4 @@ AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
 LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
 OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN
 THE SOFTWARE.
+

lib/composer/composer/autoload_classmap.php

@@ -319,6 +319,7 @@
     'OCP\\DB\\IPreparedStatement' => $baseDir . '/lib/public/DB/IPreparedStatement.php',
     'OCP\\DB\\IResult' => $baseDir . '/lib/public/DB/IResult.php',
     'OCP\\DB\\ISchemaWrapper' => $baseDir . '/lib/public/DB/ISchemaWrapper.php',
+    'OCP\\DB\\QueryBuilder\\ConflictResolutionMode' => $baseDir . '/lib/public/DB/QueryBuilder/ConflictResolutionMode.php',
     'OCP\\DB\\QueryBuilder\\ICompositeExpression' => $baseDir . '/lib/public/DB/QueryBuilder/ICompositeExpression.php',
     'OCP\\DB\\QueryBuilder\\IExpressionBuilder' => $baseDir . '/lib/public/DB/QueryBuilder/IExpressionBuilder.php',
     'OCP\\DB\\QueryBuilder\\IFunctionBuilder' => $baseDir . '/lib/public/DB/QueryBuilder/IFunctionBuilder.php',

lib/composer/composer/autoload_static.php

@@ -360,6 +360,7 @@ class ComposerStaticInit749170dad3f5e7f9ca158f5a9f04f6a2
         'OCP\\DB\\IPreparedStatement' => __DIR__ . '/../../..' . '/lib/public/DB/IPreparedStatement.php',
         'OCP\\DB\\IResult' => __DIR__ . '/../../..' . '/lib/public/DB/IResult.php',
         'OCP\\DB\\ISchemaWrapper' => __DIR__ . '/../../..' . '/lib/public/DB/ISchemaWrapper.php',
+        'OCP\\DB\\QueryBuilder\\ConflictResolutionMode' => __DIR__ . '/../../..' . '/lib/public/DB/QueryBuilder/ConflictResolutionMode.php',
         'OCP\\DB\\QueryBuilder\\ICompositeExpression' => __DIR__ . '/../../..' . '/lib/public/DB/QueryBuilder/ICompositeExpression.php',
         'OCP\\DB\\QueryBuilder\\IExpressionBuilder' => __DIR__ . '/../../..' . '/lib/public/DB/QueryBuilder/IExpressionBuilder.php',
         'OCP\\DB\\QueryBuilder\\IFunctionBuilder' => __DIR__ . '/../../..' . '/lib/public/DB/QueryBuilder/IFunctionBuilder.php',

lib/private/DB/QueryBuilder/ExtendedQueryBuilder.php

@@ -9,6 +9,7 @@
 namespace OC\DB\QueryBuilder;
 
 use OCP\DB\IResult;
+use OCP\DB\QueryBuilder\ConflictResolutionMode;
 use OCP\DB\QueryBuilder\IQueryBuilder;
 use OCP\IDBConnection;
 
@@ -290,4 +291,9 @@ public function getOutputColumns(): array {
 	public function prefixTableName(string $table): string {
 		return $this->builder->prefixTableName($table);
 	}
+
+	public function forUpdate(ConflictResolutionMode $conflictResolutionMode = ConflictResolutionMode::Ordinary): self {
+		$this->builder->forUpdate($conflictResolutionMode);
+		return $this;
+	}
 }

lib/private/DB/QueryBuilder/QueryBuilder.php

@@ -19,6 +19,7 @@
 use OC\DB\QueryBuilder\FunctionBuilder\SqliteFunctionBuilder;
 use OC\SystemConfig;
 use OCP\DB\IResult;
+use OCP\DB\QueryBuilder\ConflictResolutionMode;
 use OCP\DB\QueryBuilder\ICompositeExpression;
 use OCP\DB\QueryBuilder\IExpressionBuilder;
 use OCP\DB\QueryBuilder\IFunctionBuilder;
@@ -27,6 +28,7 @@
 use OCP\DB\QueryBuilder\IQueryBuilder;
 use OCP\DB\QueryBuilder\IQueryFunction;
 use OCP\IDBConnection;
+use Override;
 use Psr\Log\LoggerInterface;
 
 class QueryBuilder implements IQueryBuilder {
@@ -1364,4 +1366,12 @@ public function runAcrossAllShards(): self {
 		return $this;
 	}
 
+	#[Override]
+	public function forUpdate(ConflictResolutionMode $conflictResolutionMode = ConflictResolutionMode::Ordinary): self {
+		match ($conflictResolutionMode) {
+			ConflictResolutionMode::Ordinary => $this->queryBuilder->forUpdate(),
+			ConflictResolutionMode::SkipLocked => $this->queryBuilder->forUpdate(\Doctrine\DBAL\Query\ForUpdate\ConflictResolutionMode::SKIP_LOCKED),
+		};
+		return $this;
+	}
 }

lib/private/Files/Cache/Propagator.php

@@ -11,6 +11,7 @@
 
 use OC\DB\Exceptions\DbalException;
 use OC\Files\Storage\Wrapper\Encryption;
+use OCP\DB\QueryBuilder\ILiteral;
 use OCP\DB\QueryBuilder\IQueryBuilder;
 use OCP\Files\Cache\IPropagator;
 use OCP\Files\Storage\IReliableEtagStorage;
@@ -59,12 +60,11 @@ public function propagateChange(string $internalPath, int $time, int $sizeDiffer
 		}
 
 		$parentHashes = array_map('md5', $parents);
+		sort($parentHashes); // Ensure rows are always locked in the same order
 		$etag = uniqid(); // since we give all folders the same etag we don't ask the storage for the etag
 
 		$builder = $this->connection->getQueryBuilder();
-		$hashParams = array_map(function ($hash) use ($builder) {
-			return $builder->expr()->literal($hash);
-		}, $parentHashes);
+		$hashParams = array_map(static fn (string $hash): ILiteral => $builder->expr()->literal($hash), $parentHashes);
 
 		$builder->update('filecache')
 			->set('mtime', $builder->func()->greatest('mtime', $builder->createNamedParameter($time, IQueryBuilder::PARAM_INT)))
@@ -105,14 +105,31 @@ public function propagateChange(string $internalPath, int $time, int $sizeDiffer
 
 		for ($i = 0; $i < self::MAX_RETRIES; $i++) {
 			try {
-				$builder->executeStatement();
+				if ($this->connection->getDatabaseProvider() !== IDBConnection::PLATFORM_SQLITE) {
+					$this->connection->beginTransaction();
+					// Lock all the rows first with a SELECT FOR UPDATE ordered by path_hash
+					$forUpdate = $this->connection->getQueryBuilder();
+					$forUpdate->select('fileid')
+						->from('filecache')
+						->where($forUpdate->expr()->eq('storage', $forUpdate->createNamedParameter($storageId, IQueryBuilder::PARAM_INT)))
+						->andWhere($forUpdate->expr()->in('path_hash', $hashParams))
+						->orderBy('path_hash')
+						->forUpdate()
+						->executeQuery();
+					$builder->executeStatement();
+					$this->connection->commit();
+				} else {
+					$builder->executeStatement();
+				}
 				break;
 			} catch (DbalException $e) {
+				if ($this->connection->getDatabaseProvider() !== IDBConnection::PLATFORM_SQLITE) {
+					$this->connection->rollBack();
+				}
 				if (!$e->isRetryable()) {
 					throw $e;
 				}
 
-				/** @var LoggerInterface $loggerInterface */
 				$loggerInterface = Server::get(LoggerInterface::class);
 				$loggerInterface->warning('Retrying propagation query after retryable exception.', [ 'exception' => $e ]);
 			}
@@ -160,36 +177,87 @@ public function commitBatch(): void {
 		}
 		$this->inBatch = false;
 
+		// Ensure rows are always locked in the same order
+		uasort($this->batch, static fn (array $a, array $b) => $a['hash'] <=> $b['hash']);
+
 		try {
 			$this->connection->beginTransaction();
 
-			$query = $this->connection->getQueryBuilder();
 			$storageId = $this->storage->getCache()->getNumericStorageId();
 
-			$query->update('filecache')
-				->set('mtime', $query->func()->greatest('mtime', $query->createParameter('time')))
-				->set('etag', $query->expr()->literal(uniqid()))
-				->where($query->expr()->eq('storage', $query->createNamedParameter($storageId, IQueryBuilder::PARAM_INT)))
-				->andWhere($query->expr()->eq('path_hash', $query->createParameter('hash')));
-
-			$sizeQuery = $this->connection->getQueryBuilder();
-			$sizeQuery->update('filecache')
-				->set('size', $sizeQuery->func()->add('size', $sizeQuery->createParameter('size')))
-				->where($query->expr()->eq('storage', $sizeQuery->createNamedParameter($storageId, IQueryBuilder::PARAM_INT)))
-				->andWhere($query->expr()->eq('path_hash', $sizeQuery->createParameter('hash')))
-				->andWhere($sizeQuery->expr()->gt('size', $sizeQuery->createNamedParameter(-1, IQueryBuilder::PARAM_INT)));
-
-			foreach ($this->batch as $item) {
-				$query->setParameter('time', $item['time'], IQueryBuilder::PARAM_INT);
-				$query->setParameter('hash', $item['hash']);
-
-				$query->executeStatement();
-
-				if ($item['size']) {
-					$sizeQuery->setParameter('size', $item['size'], IQueryBuilder::PARAM_INT);
-					$sizeQuery->setParameter('hash', $item['hash']);
-
-					$sizeQuery->executeStatement();
+			if ($this->connection->getDatabaseProvider() !== IDBConnection::PLATFORM_SQLITE) {
+				// Lock the rows before updating then with a SELECT FOR UPDATE
+				// The select also allow us to fetch the fileid and then use these in the UPDATE
+				// queries as a faster lookup than the path_hash
+				$hashes = array_map(static fn (array $a): string => $a['hash'], $this->batch);
+
+				foreach (array_chunk($hashes, 1000) as $hashesChunk) {
+					$query = $this->connection->getQueryBuilder();
+					$result = $query->select('fileid', 'path', 'path_hash', 'size')
+						->from('filecache')
+						->where($query->expr()->eq('storage', $query->createNamedParameter($storageId, IQueryBuilder::PARAM_INT)))
+						->andWhere($query->expr()->in('path_hash', $query->createNamedParameter($hashesChunk, IQueryBuilder::PARAM_STR_ARRAY)))
+						->orderBy('path_hash')
+						->forUpdate()
+						->executeQuery();
+
+					$query = $this->connection->getQueryBuilder();
+					$query->update('filecache')
+						->set('mtime', $query->func()->greatest('mtime', $query->createParameter('time')))
+						->set('etag', $query->expr()->literal(uniqid()))
+						->where($query->expr()->eq('storage', $query->createNamedParameter($storageId, IQueryBuilder::PARAM_INT)))
+						->andWhere($query->expr()->eq('fileid', $query->createParameter('fileid')));
+
+					$queryWithSize = $this->connection->getQueryBuilder();
+					$queryWithSize->update('filecache')
+						->set('mtime', $queryWithSize->func()->greatest('mtime', $queryWithSize->createParameter('time')))
+						->set('etag', $queryWithSize->expr()->literal(uniqid()))
+						->set('size', $queryWithSize->func()->add('size', $queryWithSize->createParameter('size')))
+						->where($queryWithSize->expr()->eq('storage', $queryWithSize->createNamedParameter($storageId, IQueryBuilder::PARAM_INT)))
+						->andWhere($queryWithSize->expr()->eq('fileid', $queryWithSize->createParameter('fileid')));
+
+					while ($row = $result->fetchAssociative()) {
+						$item = $this->batch[$row['path']];
+						if ($item['size'] && $row['size'] > -1) {
+							$queryWithSize->setParameter('fileid', $row['fileid'], IQueryBuilder::PARAM_INT)
+								->setParameter('size', $item['size'], IQueryBuilder::PARAM_INT)
+								->setParameter('time', $item['time'], IQueryBuilder::PARAM_INT)
+								->executeStatement();
+						} else {
+							$query->setParameter('fileid', $row['fileid'], IQueryBuilder::PARAM_INT)
+								->setParameter('time', $item['time'], IQueryBuilder::PARAM_INT)
+								->executeStatement();
+						}
+					}
+				}
+			} else {
+				// No FOR UPDATE support in Sqlite, but instead the whole table is locked
+				$query = $this->connection->getQueryBuilder();
+				$query->update('filecache')
+					->set('mtime', $query->func()->greatest('mtime', $query->createParameter('time')))
+					->set('etag', $query->expr()->literal(uniqid()))
+					->where($query->expr()->eq('storage', $query->createNamedParameter($storageId, IQueryBuilder::PARAM_INT)))
+					->andWhere($query->expr()->eq('path_hash', $query->createParameter('hash')));
+
+				$queryWithSize = $this->connection->getQueryBuilder();
+				$queryWithSize->update('filecache')
+					->set('mtime', $queryWithSize->func()->greatest('mtime', $queryWithSize->createParameter('time')))
+					->set('etag', $queryWithSize->expr()->literal(uniqid()))
+					->set('size', $queryWithSize->func()->add('size', $queryWithSize->createParameter('size')))
+					->where($queryWithSize->expr()->eq('storage', $queryWithSize->createNamedParameter($storageId, IQueryBuilder::PARAM_INT)))
+					->andWhere($queryWithSize->expr()->eq('path_hash', $queryWithSize->createParameter('hash')));
+
+				foreach ($this->batch as $item) {
+					if ($item['size']) {
+						$queryWithSize->setParameter('hash', $item['hash'], IQueryBuilder::PARAM_STR)
+							->setParameter('time', $item['time'], IQueryBuilder::PARAM_INT)
+							->setParameter('size', $item['size'], IQueryBuilder::PARAM_INT)
+							->executeStatement();
+					} else {
+						$query->setParameter('hash', $item['hash'], IQueryBuilder::PARAM_STR)
+							->setParameter('time', $item['time'], IQueryBuilder::PARAM_INT)
+							->executeStatement();
+					}
 				}
 			}
 

lib/public/DB/QueryBuilder/ConflictResolutionMode.php

@@ -0,0 +1,26 @@
+<?php
+
+declare(strict_types=1);
+
+/**
+ * SPDX-FileCopyrightText: 2026 Nextcloud GmbH and Nextcloud contributors
+ * SPDX-License-Identifier: AGPL-3.0-or-later
+ */
+
+namespace OCP\DB\QueryBuilder;
+
+/**
+ * Conflict resolution mode for "FOR UPDATE" select queries.
+ *
+ * @since 34.0.0
+ */
+enum ConflictResolutionMode {
+	/**
+	 * Wait for the row to be unlocked.
+	 */
+	case Ordinary;
+	/**
+	 * Skip the row if it is locked.
+	 */
+	case SkipLocked;
+}

lib/public/DB/QueryBuilder/IQueryBuilder.php

@@ -1093,4 +1093,12 @@ public function runAcrossAllShards(): self;
 	 * @since 30.0.0
 	 */
 	public function getOutputColumns(): array;
+
+	/**
+	 * Locks the queried rows for a subsequent update.
+	 *
+	 * @return $this
+	 * @since 34.0.0
+	 */
+	public function forUpdate(ConflictResolutionMode $conflictResolutionMode = ConflictResolutionMode::Ordinary): self;
 }

tests/lib/Files/Cache/PropagatorTest.php

@@ -15,8 +15,7 @@
 
 #[\PHPUnit\Framework\Attributes\Group('DB')]
 class PropagatorTest extends TestCase {
-	/** @var IStorage */
-	private $storage;
+	private IStorage $storage;
 
 	protected function setUp(): void {
 		parent::setUp();