diff --git a/src/ContentRepository.PostgreSql/Data/PgSqlDataProvider.cs b/src/ContentRepository.PostgreSql/Data/PgSqlDataProvider.cs index 353e962f8..7853af2f1 100644 --- a/src/ContentRepository.PostgreSql/Data/PgSqlDataProvider.cs +++ b/src/ContentRepository.PostgreSql/Data/PgSqlDataProvider.cs @@ -295,6 +295,142 @@ await ExecuteEmbeddedNonQueryScriptAsync( private IBlobStorage BlobStorage => Providers.Instance.BlobStorage; + public override async System.Threading.Tasks.Task DeleteNodeAsync(NodeHeadData nodeHeadData, int partitionSize, + CancellationToken cancellationToken) + { + try + { + var effectivePartitionSize = partitionSize > 0 ? partitionSize : 500; + using var op = SnTrace.Database.StartOperation("PgSqlDataProvider: " + + "DeleteNode: NodeId: {0}, Path: {1}, partitionSize: {2}", + nodeHeadData.NodeId, nodeHeadData.Path, effectivePartitionSize); + + using (var ctx = CreateDataContext(cancellationToken)) + { + var transactionTimeout = TimeSpan.FromSeconds(_dataOptions?.LongTransactionTimeout + ?? new DataOptions().LongTransactionTimeout); + using var transaction = ctx.BeginTransaction(timeout: transactionTimeout); + + var nodeInfo = await ctx.ExecuteReaderAsync( + @"SELECT ""Path""::TEXT, ""Timestamp"" FROM ""Nodes"" WHERE ""NodeId"" = @NodeId", + cmd => { cmd.Parameters.Add(ctx.CreateParameter("@NodeId", DbType.Int32, nodeHeadData.NodeId)); }, + async (reader, cancel) => + { + if (!await reader.ReadAsync(cancel).ConfigureAwait(false)) + return null; + + return new + { + Path = reader.GetString(0), + Timestamp = ConvertTimestampToInt64(reader.GetValue(1)) + }; + }).ConfigureAwait(false); + + if (nodeInfo != null) + { + if (nodeInfo.Timestamp != nodeHeadData.Timestamp) + throw new NodeIsOutOfDateException( + $"Node is out of date. Id: {nodeHeadData.NodeId}, path: {nodeInfo.Path}."); + + await InitializeDeleteTempTablesAsync(ctx, nodeInfo.Path).ConfigureAwait(false); + + while (true) + { + await ctx.ExecuteNonQueryAsync(@"TRUNCATE TABLE ""DeletePartitionNodeIds""") + .ConfigureAwait(false); + + var partitionCount = Convert.ToInt32(await ctx.ExecuteScalarAsync(@" +WITH partitioned AS ( + INSERT INTO ""DeletePartitionNodeIds"" (""NodeId"") + SELECT ""NodeId"" FROM ""DeleteNodeIds"" ORDER BY ""Id"" LIMIT @PartitionSize + RETURNING 1 +) +SELECT COUNT(1) FROM partitioned; +", cmd => + { + cmd.Parameters.Add(ctx.CreateParameter("@PartitionSize", DbType.Int32, effectivePartitionSize)); + }).ConfigureAwait(false)); + + if (partitionCount == 0) + break; + + await ctx.ExecuteNonQueryAsync(@" +TRUNCATE TABLE ""DeleteVersionIds""; + +INSERT INTO ""DeleteVersionIds"" (""VersionId"") +SELECT ""VersionId"" FROM ""Versions"" +WHERE ""NodeId"" IN (SELECT ""NodeId"" FROM ""DeletePartitionNodeIds""); + +DELETE FROM ""BinaryProperties"" +WHERE ""VersionId"" IN (SELECT ""VersionId"" FROM ""DeleteVersionIds""); + +DELETE FROM ""LongTextProperties"" +WHERE ""VersionId"" IN (SELECT ""VersionId"" FROM ""DeleteVersionIds""); + +DELETE FROM ""ReferenceProperties"" +WHERE ""VersionId"" IN (SELECT ""VersionId"" FROM ""DeleteVersionIds"") + OR ""ReferredNodeId"" IN (SELECT ""NodeId"" FROM ""DeletePartitionNodeIds""); + +DELETE FROM ""Versions"" +WHERE ""NodeId"" IN (SELECT ""NodeId"" FROM ""DeletePartitionNodeIds""); + +DELETE FROM ""Nodes"" +WHERE ""NodeId"" IN (SELECT ""NodeId"" FROM ""DeletePartitionNodeIds""); + +DELETE FROM ""DeleteNodeIds"" +WHERE ""NodeId"" IN (SELECT ""NodeId"" FROM ""DeletePartitionNodeIds""); +").ConfigureAwait(false); + } + } + + transaction.Commit(); + } + + await BlobStorage.DeleteOrphanedFilesAsync(cancellationToken); + + op.Successful = true; + } + catch (DataException) + { + throw; + } + catch (Exception e) + { + const string msg = "Node was not updated. For more details see the inner exception."; + var transformedException = GetException(e, msg); + if (transformedException != null) + throw transformedException; + throw new DataException(msg, e); + } + } + + private async System.Threading.Tasks.Task InitializeDeleteTempTablesAsync(SnDataContext ctx, string path) + { + await ctx.ExecuteNonQueryAsync(@" +CREATE TEMP TABLE ""DeleteNodeIds"" ( + ""Id"" INT GENERATED ALWAYS AS IDENTITY, + ""NodeId"" INT PRIMARY KEY +) ON COMMIT DROP; + +CREATE TEMP TABLE ""DeletePartitionNodeIds"" ( + ""NodeId"" INT PRIMARY KEY +) ON COMMIT DROP; + +CREATE TEMP TABLE ""DeleteVersionIds"" ( + ""VersionId"" INT PRIMARY KEY +) ON COMMIT DROP; + +INSERT INTO ""DeleteNodeIds"" (""NodeId"") +SELECT ""NodeId"" FROM ""Nodes"" +WHERE lower(""Path""::TEXT) = lower(@Path) + OR left(lower(""Path""::TEXT), length(CAST(@Path AS TEXT)) + 1) = lower(CAST(@Path AS TEXT) || '/') +ORDER BY ""Path"" DESC; +", cmd => + { + cmd.Parameters.Add(ctx.CreateParameter("@Path", DbType.String, PathMaxLength, path)); + }).ConfigureAwait(false); + } + private async System.Threading.Tasks.Task IsDatabaseAlreadyInstalledAsync(CancellationToken cancellationToken) { try diff --git a/src/ContentRepository/Search/Indexing/IndexManager.cs b/src/ContentRepository/Search/Indexing/IndexManager.cs index 69f477b49..f73f0406c 100644 --- a/src/ContentRepository/Search/Indexing/IndexManager.cs +++ b/src/ContentRepository/Search/Indexing/IndexManager.cs @@ -358,9 +358,31 @@ internal async STT.Task AddTreeAsync(string treeRoot, int activityId, var excludedNodeTypes = GetNotIndexedNodeTypes(); var docs = _dataStore.LoadIndexDocumentsAsync(treeRoot, excludedNodeTypes) .Select(CreateIndexDocument); + docs = TraceAddTreeProgress(docs, treeRoot, activityId); await IndexingEngine.WriteIndexAsync(delTerms, null, docs, cancellationToken).ConfigureAwait(false); return true; } + + private IEnumerable TraceAddTreeProgress(IEnumerable documents, string treeRoot, + int activityId) + { + var count = 0; + SnTrace.Index.Write("LM: AddTreeActivity progress started. ActivityId:{0}, Path:{1}", + activityId, treeRoot); + + foreach (var document in documents) + { + count++; + if (count % 500 == 0) + SnTrace.Index.Write("LM: AddTreeActivity progress. ActivityId:{0}, Path:{1}, Documents:{2}", + activityId, treeRoot, count); + + yield return document; + } + + SnTrace.Index.Write("LM: AddTreeActivity progress finished. ActivityId:{0}, Path:{1}, Documents:{2}", + activityId, treeRoot, count); + } /* ==================================================================== IndexDocument management */ diff --git a/src/Storage/TreeLockController.cs b/src/Storage/TreeLockController.cs index 57a201f26..cf415e013 100644 --- a/src/Storage/TreeLockController.cs +++ b/src/Storage/TreeLockController.cs @@ -56,25 +56,89 @@ public async Task AcquireAsync(CancellationToken cancellationToken, pa SnTrace.ContentOperation.Write("TreeLock: Acquiring lock for {0}", string.Join(", ", paths)); - var lockTasks = paths.Select(p => _dataStore.AcquireTreeLockAsync(p, cancellationToken)); - var lockIds = await Task.WhenAll(lockTasks).ConfigureAwait(false); + var lockIds = new int[0]; + Task[] lockTasks = null; - for (var i = 0; i < lockIds.Length; i++) + try { - cancellationToken.ThrowIfCancellationRequested(); + lockTasks = paths.Select(p => _dataStore.AcquireTreeLockAsync(p, CancellationToken.None)).ToArray(); + lockIds = await Task.WhenAll(lockTasks).ConfigureAwait(false); - if (lockIds[i] == 0) + for (var i = 0; i < lockIds.Length; i++) { - await _dataStore.ReleaseTreeLockAsync(lockIds, cancellationToken).ConfigureAwait(false); - var msg = "Cannot acquire a tree lock for " + paths[i]; - SnTrace.ContentOperation.Write("TreeLock: " + msg); - throw new LockedTreeException(msg); + cancellationToken.ThrowIfCancellationRequested(); + + if (lockIds[i] == 0) + { + var msg = "Cannot acquire a tree lock for " + paths[i]; + await WriteBlockingLocksAsync(paths[i]).ConfigureAwait(false); + SnTrace.ContentOperation.Write("TreeLock: " + msg); + throw new LockedTreeException(msg); + } } } + catch + { + if ((lockIds == null || lockIds.Length == 0) && lockTasks != null) + lockIds = lockTasks + .Where(task => task.IsCompletedSuccessfully) + .Select(task => task.Result) + .ToArray(); + + try + { + await ReleaseLocksAsync(lockIds).ConfigureAwait(false); + } + catch (System.Exception e) + { + _logger.LogWarning(e, "TreeLock cleanup failed after an unsuccessful acquire."); + } + throw; + } var logOp = SnTrace.ContentOperation.StartOperation("TreeLock: {0} for {1}", lockIds, paths); return new TreeLock(logOp, _dataStore, lockIds); } + + private async Task ReleaseLocksAsync(IEnumerable lockIds) + { + var existingLockIds = lockIds?.Where(id => id != 0).ToArray(); + if (existingLockIds == null || existingLockIds.Length == 0) + return; + + await _dataStore.ReleaseTreeLockAsync(existingLockIds, CancellationToken.None).ConfigureAwait(false); + } + + private async Task WriteBlockingLocksAsync(string path) + { + try + { + var locks = await _dataStore.LoadAllTreeLocksAsync(CancellationToken.None).ConfigureAwait(false); + var blockingLocks = locks + .Where(x => IsBlocking(path, x.Value)) + .Select(x => $"{x.Key}:{x.Value}") + .ToArray(); + + SnTrace.ContentOperation.Write("TreeLock: Blocking locks for {0}: {1}", + path, blockingLocks.Length == 0 ? "[none]" : string.Join(", ", blockingLocks)); + } + catch (System.Exception e) + { + _logger.LogWarning(e, "Could not load blocking tree locks for {Path}.", path); + } + } + + private static bool IsBlocking(string requestedPath, string lockedPath) + { + return IsSameOrAncestor(requestedPath, lockedPath) || IsSameOrAncestor(lockedPath, requestedPath); + } + + private static bool IsSameOrAncestor(string path, string ancestorPath) + { + return path.Equals(ancestorPath, System.StringComparison.OrdinalIgnoreCase) || + path.StartsWith(ancestorPath + "/", System.StringComparison.OrdinalIgnoreCase); + } + public async Task AssertFreeAsync(CancellationToken cancellationToken, params string[] paths) { SnTrace.ContentOperation.Write("TreeLock: Checking {0}", string.Join(", ", paths)); diff --git a/src/Tests/SenseNet.ContentRepository.Tests/TreeLockTests.cs b/src/Tests/SenseNet.ContentRepository.Tests/TreeLockTests.cs index da1f96aa4..54edcdf77 100644 --- a/src/Tests/SenseNet.ContentRepository.Tests/TreeLockTests.cs +++ b/src/Tests/SenseNet.ContentRepository.Tests/TreeLockTests.cs @@ -230,6 +230,60 @@ await Test(async () => }); } + [TestMethod] + public async STT.Task TreeLock_CanceledAcquireReleasesLocks() + { + await Test(async () => + { + var locks = await TreeLock.GetAllLocksAsync(CancellationToken.None); + Assert.AreEqual(0, locks.Count); + + using var cts = new CancellationTokenSource(); + var acquireTask = TreeLock.AcquireAsync(cts.Token, "/Root/A/B/C"); + cts.Cancel(); + + try + { + using (await acquireTask.ConfigureAwait(false)) + { + // The acquire may complete before the cancellation is observed. + } + } + catch (System.OperationCanceledException) + { + // expected when cancellation wins the race + } + + locks = await TreeLock.GetAllLocksAsync(CancellationToken.None); + Assert.AreEqual(0, locks.Count); + }); + } + + [TestMethod] + public async STT.Task TreeLock_PartialAcquireFailureReleasesAcquiredLocks() + { + await Test(async () => + { + using (await TreeLock.AcquireAsync(CancellationToken.None, "/Root/A/B/C")) + { + try + { + await TreeLock.AcquireAsync(CancellationToken.None, "/Root/X", "/Root/A/B/C/D"); + Assert.Fail("LockedTreeException was not thrown."); + } + catch (LockedTreeException) + { + // expected + } + + var locks = await TreeLock.GetAllLocksAsync(CancellationToken.None); + Assert.AreEqual(1, locks.Count); + Assert.IsTrue(locks.ContainsValue("/Root/A/B/C")); + Assert.IsFalse(locks.ContainsValue("/Root/X")); + } + }); + } + private bool IsLocked(string path) { try