diff --git a/CLAUDE.md b/CLAUDE.md index 50d5822..46247ef 100644 --- a/CLAUDE.md +++ b/CLAUDE.md @@ -48,6 +48,7 @@ This is a .NET library (`ktsu.Essentials`) providing high-performance interfaces - `Shared/SpecialFunctions.cs` - Incomplete gamma, incomplete beta, error function, normal quantile and cancellation-free `ExpM1`/`Log1P`, linked into the Normal, LogNormal, Exponential, Binomial, Poisson and Geometric providers rather than placed in the interfaces-only package - `Shared/DistributionArguments.cs` - Parameter validation shared by every distribution provider, linked in the same way - `Shared/PersistenceDefaults.cs` - The `IPersistenceProvider` default `RetrieveOrCreateAsync` body, linked into the FileSystem and Temp providers, which declare that member themselves and so hide the default +- `Shared/AtomicFile.cs` - The FileSystem and Temp providers' write and read paths: each store stages its data in a uniquely named `.tmp` beside the target and publishes it with an overwriting move, retrying briefly while Windows holds the target open, so concurrent writers can't collide and a reader never sees the key missing; reads retry only the Windows sharing violation a replace in progress raises. Linked into both providers like `PersistenceDefaults.cs` - `Essentials/ISerializationProvider.cs` - Object serialization/deserialization interface - `Essentials/ISerializationOptions.cs` - Configurable serialization options (naming, inclusion, boxing policies) - `Essentials/ICacheProvider.cs` - Generic cache interface with expiration and get-or-add diff --git a/Essentials.PersistenceProviders.FileSystem/Essentials.PersistenceProviders.FileSystem.csproj b/Essentials.PersistenceProviders.FileSystem/Essentials.PersistenceProviders.FileSystem.csproj index 2f1340e..4e5bf75 100644 --- a/Essentials.PersistenceProviders.FileSystem/Essentials.PersistenceProviders.FileSystem.csproj +++ b/Essentials.PersistenceProviders.FileSystem/Essentials.PersistenceProviders.FileSystem.csproj @@ -16,6 +16,7 @@ + diff --git a/Essentials.PersistenceProviders.FileSystem/FileSystemPersistenceProvider.cs b/Essentials.PersistenceProviders.FileSystem/FileSystemPersistenceProvider.cs index c89b913..a3b5b18 100644 --- a/Essentials.PersistenceProviders.FileSystem/FileSystemPersistenceProvider.cs +++ b/Essentials.PersistenceProviders.FileSystem/FileSystemPersistenceProvider.cs @@ -68,16 +68,7 @@ public async Task StoreAsync(TKey key, T obj, CancellationToken cancellationT _fileSystemProvider.Directory.CreateDirectory(directory); } - // Write to temporary file first, then move for atomic operation - string tempFilePath = filePath + ".tmp"; - await _fileSystemProvider.File.WriteAllTextAsync(tempFilePath, serializedData, cancellationToken).ConfigureAwait(false); - - // Atomic move - if (_fileSystemProvider.File.Exists(filePath)) - { - _fileSystemProvider.File.Delete(filePath); - } - _fileSystemProvider.File.Move(tempFilePath, filePath); + await AtomicFile.WriteAllTextAsync(_fileSystemProvider, filePath, serializedData, cancellationToken).ConfigureAwait(false); } catch (Exception ex) { @@ -102,7 +93,7 @@ public async Task StoreAsync(TKey key, T obj, CancellationToken cancellationT return default; } - string serializedData = await _fileSystemProvider.File.ReadAllTextAsync(filePath, cancellationToken).ConfigureAwait(false); + string serializedData = await AtomicFile.ReadAllTextAsync(_fileSystemProvider, filePath, cancellationToken).ConfigureAwait(false); if (string.IsNullOrEmpty(serializedData)) { diff --git a/Essentials.PersistenceProviders.Temp/Essentials.PersistenceProviders.Temp.csproj b/Essentials.PersistenceProviders.Temp/Essentials.PersistenceProviders.Temp.csproj index 2f1340e..4e5bf75 100644 --- a/Essentials.PersistenceProviders.Temp/Essentials.PersistenceProviders.Temp.csproj +++ b/Essentials.PersistenceProviders.Temp/Essentials.PersistenceProviders.Temp.csproj @@ -16,6 +16,7 @@ + diff --git a/Essentials.PersistenceProviders.Temp/TempPersistenceProvider.cs b/Essentials.PersistenceProviders.Temp/TempPersistenceProvider.cs index 3649650..d668b15 100644 --- a/Essentials.PersistenceProviders.Temp/TempPersistenceProvider.cs +++ b/Essentials.PersistenceProviders.Temp/TempPersistenceProvider.cs @@ -65,16 +65,7 @@ public async Task StoreAsync(TKey key, T obj, CancellationToken cancellationT _fileSystemProvider.Directory.CreateDirectory(directory); } - // Write to temporary file first, then move for atomic operation - string tempFilePath = filePath + ".tmp"; - await _fileSystemProvider.File.WriteAllTextAsync(tempFilePath, serializedData, cancellationToken).ConfigureAwait(false); - - // Atomic move - if (_fileSystemProvider.File.Exists(filePath)) - { - _fileSystemProvider.File.Delete(filePath); - } - _fileSystemProvider.File.Move(tempFilePath, filePath); + await AtomicFile.WriteAllTextAsync(_fileSystemProvider, filePath, serializedData, cancellationToken).ConfigureAwait(false); } catch (Exception ex) { @@ -99,7 +90,7 @@ public async Task StoreAsync(TKey key, T obj, CancellationToken cancellationT return default; } - string serializedData = await _fileSystemProvider.File.ReadAllTextAsync(filePath, cancellationToken).ConfigureAwait(false); + string serializedData = await AtomicFile.ReadAllTextAsync(_fileSystemProvider, filePath, cancellationToken).ConfigureAwait(false); if (string.IsNullOrEmpty(serializedData)) { diff --git a/Essentials.Tests/PersistenceProviderTests.cs b/Essentials.Tests/PersistenceProviderTests.cs index d2f01e6..678f210 100644 --- a/Essentials.Tests/PersistenceProviderTests.cs +++ b/Essentials.Tests/PersistenceProviderTests.cs @@ -686,6 +686,80 @@ public sealed class CyclicNode public CyclicNode? Next { get; set; } } + [TestMethod] + public async Task Concurrent_Stores_To_One_Key_Never_Throw_Or_Hide_The_Key() + { + // Every writer staged its data in the same ".tmp", so overlapping writers moved each other's file + // away and threw, and the delete-then-move left a window in which a reader saw the key as missing + // (issue #66). + string fileSystemDir = Path.Join(Path.GetTempPath(), "PersistenceTests_FS_" + Guid.NewGuid().ToString("N")[..8]); + NativeFileSystemProvider fs = new(); + JsonSerializationProvider serializer = new(); + using TempPersistenceProvider temp = new(fs, serializer, "PersistenceTests_Temp_" + Guid.NewGuid().ToString("N")[..8]); + + try + { + IPersistenceProvider[] providers = + [ + new FileSystemPersistenceProvider(fs, serializer, fileSystemDir), + temp, + ]; + + foreach (IPersistenceProvider persistence in providers) + { + string name = persistence.ProviderName; + const int writers = 200; + + await persistence.StoreAsync("k", new TestData { Name = "initial" }, TestContext.CancellationToken).ConfigureAwait(false); + + using CancellationTokenSource stopReading = new(); + int missing = 0; + Task reader = Task.Run(async () => + { + while (!stopReading.IsCancellationRequested) + { + if (!await persistence.ExistsAsync("k", TestContext.CancellationToken).ConfigureAwait(false) + || await persistence.RetrieveAsync("k", TestContext.CancellationToken).ConfigureAwait(false) is null) + { + Interlocked.Increment(ref missing); + } + } + }, TestContext.CancellationToken); + + Task[] stores = [.. Enumerable.Range(0, writers).Select(i => Task.Run( + () => persistence.StoreAsync("k", new TestData { Name = $"writer {i}" }, TestContext.CancellationToken), + TestContext.CancellationToken))]; + + try + { + await Task.WhenAll(stores).ConfigureAwait(false); + } + finally + { + await stopReading.CancelAsync().ConfigureAwait(false); + await reader.ConfigureAwait(false); + } + + Assert.AreEqual(0, missing, $"{name} reported a stored key as missing during concurrent writes"); + + TestData? final = await persistence.RetrieveAsync("k", TestContext.CancellationToken).ConfigureAwait(false); + Assert.IsNotNull(final); + Assert.Contains(final.Name, Enumerable.Range(0, writers).Select(i => $"writer {i}"), $"{name} published a payload no writer stored"); + + string[] keys = [.. await persistence.GetAllKeysAsync(TestContext.CancellationToken).ConfigureAwait(false)]; + Assert.HasCount(1, keys, $"{name} listed a staging file as a key"); + Assert.AreEqual("k", keys[0]); + } + + Assert.IsEmpty(Directory.GetFiles(fileSystemDir, "*.tmp"), "A staging file was left behind"); + } + finally + { + CleanupDirectory(fileSystemDir); + temp.CleanupDirectory(); + } + } + public record struct ValueData() { public int Value { get; set; } = 50; diff --git a/Shared/AtomicFile.cs b/Shared/AtomicFile.cs new file mode 100644 index 0000000..848c795 --- /dev/null +++ b/Shared/AtomicFile.cs @@ -0,0 +1,128 @@ +// Copyright (c) 2023-2026 ktsu-dev contributors + +namespace ktsu.Essentials; + +using System; +using System.IO; +using System.IO.Abstractions; +using System.Threading; +using System.Threading.Tasks; + +/// +/// Replaces a file's content in one step, so a reader sees either the old content or the new, never neither, +/// and reads it back without tripping over a replace in progress. +/// +/// +/// Linked into the FileSystem and Temp providers rather than placed in the interfaces package, following +/// PersistenceDefaults. Each write stages its content in a temporary file of its own, so overlapping +/// writers to one key cannot move each other's file away, and publishes it with an overwriting move rather +/// than a delete followed by a move, so the target never disappears in between (issue #66). +/// +internal static class AtomicFile +{ + private const int MaxAttempts = 50; + + // Win32 ERROR_SHARING_VIOLATION and ERROR_LOCK_VIOLATION, carried in the low word of IOException.HResult. + private const int SharingViolation = 32; + private const int LockViolation = 33; + + /// + /// Writes to , replacing any existing file atomically. + /// + /// The filesystem to write through. + /// The file to create or replace. + /// The text to write. + /// A token to cancel the operation. + /// A task that completes once the new content is in place. + internal static async Task WriteAllTextAsync(IFileSystem fileSystem, string filePath, string content, CancellationToken cancellationToken) + { + // Same directory as the target, so the move is a rename on one volume. The name is not derived from + // the target's, because PersistenceProviderUtilities budgets a key's filename for only the extension + // and ".tmp", and a longer suffix on a long key would exceed the 255-byte name limit. The .tmp + // extension keeps the staging file out of the providers' "*" key listings. + string? directory = fileSystem.Path.GetDirectoryName(filePath); + string tempFileName = $"{Guid.NewGuid():N}.tmp"; + string tempFilePath = string.IsNullOrEmpty(directory) ? tempFileName : fileSystem.Path.Combine(directory, tempFileName); + + try + { + await fileSystem.File.WriteAllTextAsync(tempFilePath, content, cancellationToken).ConfigureAwait(false); + + // Every attempt but the last may be refused and retried; the last one's failure surfaces. + for (int attempt = 1; attempt < MaxAttempts; attempt++) + { + try + { + Replace(fileSystem, tempFilePath, filePath); + return; + } + catch (Exception ex) when (ex is IOException or UnauthorizedAccessException) + { + // Windows refuses to replace a file another handle has open without delete sharing, which + // a concurrent reader or a second writer's replace holds for a moment. Give it that moment. + await Task.Delay(Backoff(attempt), cancellationToken).ConfigureAwait(false); + } + } + + Replace(fileSystem, tempFilePath, filePath); + } + finally + { + if (fileSystem.File.Exists(tempFilePath)) + { + fileSystem.File.Delete(tempFilePath); + } + } + } + + /// + /// Reads , waiting out a concurrent replacing it. + /// + /// The filesystem to read through. + /// The file to read. + /// A token to cancel the operation. + /// The file's content. + internal static async Task ReadAllTextAsync(IFileSystem fileSystem, string filePath, CancellationToken cancellationToken) + { + for (int attempt = 1; attempt < MaxAttempts; attempt++) + { + try + { + return await fileSystem.File.ReadAllTextAsync(filePath, cancellationToken).ConfigureAwait(false); + } + catch (IOException ex) when (IsSharingViolation(ex)) + { + // Windows holds the target exclusively for the instant an overwriting move swaps it, so a + // read that lands then is refused rather than served either version. Only that refusal is + // retried; a missing file and every other error surface at once. + await Task.Delay(Backoff(attempt), cancellationToken).ConfigureAwait(false); + } + } + + return await fileSystem.File.ReadAllTextAsync(filePath, cancellationToken).ConfigureAwait(false); + } + +#if NETSTANDARD2_1 + private static void Replace(IFileSystem fileSystem, string tempFilePath, string filePath) + { + // No overwriting Move here. Replace swaps an existing target in one step; a target that does not exist + // yet takes a plain Move, and one created by another writer in between makes that Move throw, so the + // retry takes the Replace path. + if (fileSystem.File.Exists(filePath)) + { + fileSystem.File.Replace(tempFilePath, filePath, destinationBackupFileName: null); + } + else + { + fileSystem.File.Move(tempFilePath, filePath); + } + } +#else + private static void Replace(IFileSystem fileSystem, string tempFilePath, string filePath) + => fileSystem.File.Move(tempFilePath, filePath, overwrite: true); +#endif + + private static bool IsSharingViolation(IOException ex) => (ex.HResult & 0xFFFF) is SharingViolation or LockViolation; + + private static int Backoff(int attempt) => Math.Min(attempt * 2, 20); +}