diff --git a/src/EventStore.Core.Tests/TransactionLog/LocalArchiveStorage.cs b/src/EventStore.Core.Tests/TransactionLog/LocalArchiveStorage.cs index b2dcc29ad..b5446ac5e 100644 --- a/src/EventStore.Core.Tests/TransactionLog/LocalArchiveStorage.cs +++ b/src/EventStore.Core.Tests/TransactionLog/LocalArchiveStorage.cs @@ -29,7 +29,9 @@ public ValueTask GetCheckpoint(CancellationToken ct) } Span buffer = stackalloc byte[sizeof(long)]; - using var handle = File.OpenHandle(checkpointPath); + using var handle = File.OpenHandle( + checkpointPath, + share: FileShare.ReadWrite | FileShare.Delete); if (RandomAccess.Read(handle, buffer, fileOffset: 0L) != buffer.Length) { throw new EndOfStreamException(); @@ -115,9 +117,25 @@ public async IAsyncEnumerable ListChunks([EnumeratorCancellation] Cancel public ValueTask SetCheckpoint(long checkpoint, CancellationToken ct) { var checkpointPath = Path.Combine(archivePath, archiveCheckpointFile); + var checkpointTempPath = $"{checkpointPath}.{Guid.NewGuid():N}.tmp"; Span buffer = stackalloc byte[sizeof(long)]; BinaryPrimitives.WriteInt64LittleEndian(buffer, checkpoint); - File.WriteAllBytes(checkpointPath, buffer.ToArray()); + try + { + File.WriteAllBytes(checkpointTempPath, buffer.ToArray()); + if (File.Exists(checkpointPath)) + { + File.Replace(checkpointTempPath, checkpointPath, destinationBackupFileName: null); + } + else + { + File.Move(checkpointTempPath, checkpointPath); + } + } + finally + { + File.Delete(checkpointTempPath); + } return ValueTask.FromResult(true); } diff --git a/src/EventStore.Core.XUnit.Tests/Services/Archive/Storage/LocalArchiveStorageTests.cs b/src/EventStore.Core.XUnit.Tests/Services/Archive/Storage/LocalArchiveStorageTests.cs new file mode 100644 index 000000000..71b8ce8a3 --- /dev/null +++ b/src/EventStore.Core.XUnit.Tests/Services/Archive/Storage/LocalArchiveStorageTests.cs @@ -0,0 +1,90 @@ +using System; +using System.IO; +using System.Linq; +using System.Threading; +using System.Threading.Tasks; +using EventStore.Core.Services.Archive.Naming; +using EventStore.Core.Tests.TransactionLog; +using EventStore.Core.TransactionLog.FileNamingStrategy; +using Xunit; + +namespace EventStore.Core.XUnit.Tests.Services.Archive.Storage; + +public class LocalArchiveStorageTests : DirectoryPerTest +{ + private const string ArchiveCheckpointFile = "archive.chk"; + private const int CheckpointUpdates = 2_000; + + [Fact] + public async Task checkpoint_can_be_replaced_while_a_reader_is_open() + { + var archive = new LocalArchiveStorage( + Fixture.Directory, + new ArchiveChunkNamer(new VersionedPatternFileNamingStrategy(Fixture.Directory, "chunk-")), + ArchiveCheckpointFile); + await archive.SetCheckpoint(1, CancellationToken.None); + + using var handle = File.OpenHandle( + Path.Combine(Fixture.Directory, ArchiveCheckpointFile), + share: FileShare.ReadWrite | FileShare.Delete); + await archive.SetCheckpoint(2, CancellationToken.None); + + Assert.Equal(2, await archive.GetCheckpoint(CancellationToken.None)); + } + + [Fact] + public async Task concurrent_checkpoint_reads_never_observe_partial_updates() + { + var archive = new LocalArchiveStorage( + Fixture.Directory, + new ArchiveChunkNamer(new VersionedPatternFileNamingStrategy(Fixture.Directory, "chunk-")), + ArchiveCheckpointFile); + await archive.SetCheckpoint(0, CancellationToken.None); + + var start = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously); + var firstCheckpointPublished = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously); + var checkpointObserved = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously); + var writerCompleted = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously); + var checkpointReads = 0; + var writer = Task.Run(async () => + { + await start.Task; + try + { + for (var checkpoint = 1; checkpoint <= CheckpointUpdates; checkpoint++) + { + await archive.SetCheckpoint(checkpoint, CancellationToken.None); + if (checkpoint == 1) + { + firstCheckpointPublished.SetResult(); + await checkpointObserved.Task; + } + await Task.Yield(); + } + } + finally + { + writerCompleted.SetResult(); + } + }); + + var readers = Enumerable.Range(0, Math.Clamp(Environment.ProcessorCount, 2, 8)) + .Select(_ => Task.Run(async () => + { + await start.Task; + await firstCheckpointPublished.Task; + while (!writerCompleted.Task.IsCompleted) + { + Assert.InRange(await archive.GetCheckpoint(CancellationToken.None), 0, CheckpointUpdates); + Interlocked.Increment(ref checkpointReads); + checkpointObserved.TrySetResult(); + await Task.Yield(); + } + })) + .ToArray(); + + start.SetResult(); + await Task.WhenAll(readers.Append(writer)); + Assert.True(checkpointReads > 0); + } +}