feat: add atomic file writer with SemaphoreSlim for crash-safe state writes (Gap 1.6)

- Add AtomicFileWriter static helper: writes to {path}.{random}.tmp, flushes,
  then File.Move(overwrite:true) — concurrent-safe via unique temp path per call
- Add _stateWriteLock (SemaphoreSlim 1,1) to FileStore; dispose in both Dispose
  and DisposeAsync paths
- Promote WriteStreamState to async WriteStreamStateAsync using AtomicFileWriter
  under the write lock; FlushAllPending now returns Task
- Update IStreamStore.FlushAllPending signature to Task; fix MemStore no-op impl
- Fix FileStoreCrashRecoveryTests to await FlushAllPending (3 sync→async tests)
- Add 9 AtomicFileWriterTests covering create, no-tmp-remains, overwrite,
  concurrent safety, memory overload, empty data, and large payload

src/NATS.Server/JetStream/Storage/AtomicFileWriter.cs

@@ -0,0 +1,49 @@
+namespace NATS.Server.JetStream.Storage;
+
+/// <summary>
+/// Provides crash-safe atomic file writes using the write-to-temp-then-rename pattern.
+/// Each call writes to a uniquely-named sibling temp file (<c>{path}.{random}.tmp</c>),
+/// flushes to disk, then atomically renames over the target path so that:
+/// <list type="bullet">
+///   <item>A partial write never leaves a corrupt target file.</item>
+///   <item>Concurrent callers to different paths never collide on temp files.</item>
+/// </list>
+///
+/// Reference: golang/nats-server/server/filestore.go:10599 (_writeFullState)
+/// </summary>
+public static class AtomicFileWriter
+{
+    /// <summary>
+    /// Writes <paramref name="data"/> atomically to <paramref name="path"/>.
+    /// The data is written to a unique <c>{path}.{random}.tmp</c> sibling, flushed,
+    /// then renamed over <paramref name="path"/> with overwrite semantics.
+    /// </summary>
+    public static async Task WriteAtomicallyAsync(string path, byte[] data)
+    {
+        var tmpPath = path + "." + Path.GetRandomFileName() + ".tmp";
+        await using (var fs = new FileStream(tmpPath, FileMode.CreateNew, FileAccess.Write, FileShare.None,
+                         bufferSize: 4096, useAsync: true))
+        {
+            await fs.WriteAsync(data);
+            await fs.FlushAsync();
+        }
+        File.Move(tmpPath, path, overwrite: true);
+    }
+
+    /// <summary>
+    /// Writes <paramref name="data"/> atomically to <paramref name="path"/>.
+    /// The data is written to a unique <c>{path}.{random}.tmp</c> sibling, flushed,
+    /// then renamed over <paramref name="path"/> with overwrite semantics.
+    /// </summary>
+    public static async Task WriteAtomicallyAsync(string path, ReadOnlyMemory<byte> data)
+    {
+        var tmpPath = path + "." + Path.GetRandomFileName() + ".tmp";
+        await using (var fs = new FileStream(tmpPath, FileMode.CreateNew, FileAccess.Write, FileShare.None,
+                         bufferSize: 4096, useAsync: true))
+        {
+            await fs.WriteAsync(data);
+            await fs.FlushAsync();
+        }
+        File.Move(tmpPath, path, overwrite: true);
+    }
+}

src/NATS.Server/JetStream/Storage/FileStore.cs

@@ -47,6 +47,12 @@ public sealed class FileStore : IStreamStore, IAsyncDisposable, IDisposable
     // Reference: golang/nats-server/server/filestore.go:290 (fss/ttl fields).
     private HashWheel? _ttlWheel;
 
+    // Mutual-exclusion lock for state file writes. Ensures that concurrent
+    // FlushAllPending calls (e.g. from a flush timer and a shutdown path) do not
+    // race on the stream.state / stream.state.tmp files.
+    // Reference: golang/nats-server/server/filestore.go — fsMsgBlock.mu (write lock).
+    private readonly SemaphoreSlim _stateWriteLock = new(1, 1);
+
     public int BlockCount => _blocks.Count;
     public bool UsedIndexManifestOnStartup { get; private set; }
 
@@ -662,16 +668,18 @@ public sealed class FileStore : IStreamStore, IAsyncDisposable, IDisposable
     public ValueTask DisposeAsync()
     {
         DisposeAllBlocks();
+        _stateWriteLock.Dispose();
         return ValueTask.CompletedTask;
     }
 
     /// <summary>
-    /// Synchronous dispose — releases all block file handles.
+    /// Synchronous dispose — releases all block file handles and the state-write semaphore.
     /// Allows the store to be used in synchronous test contexts with <c>using</c> blocks.
     /// </summary>
     public void Dispose()
     {
         DisposeAllBlocks();
+        _stateWriteLock.Dispose();
     }
 
     /// <summary>
@@ -1811,23 +1819,24 @@ public sealed class FileStore : IStreamStore, IAsyncDisposable, IDisposable
     /// good sequence without re-scanning every block.
     /// Reference: golang/nats-server/server/filestore.go:5783 (flushPendingWritesUnlocked).
     /// </summary>
-    public void FlushAllPending()
+    public async Task FlushAllPending()
     {
         _activeBlock?.Flush();
-        WriteStreamState();
+        await WriteStreamStateAsync();
     }
 
     /// <summary>
-    /// Atomically persists a compact stream state snapshot to disk using the
-    /// write-to-temp-then-rename pattern so that a partial write never leaves
-    /// a corrupt state file.
+    /// Atomically persists a compact stream state snapshot to disk using
+    /// <see cref="AtomicFileWriter"/> (write-to-temp-then-rename) so that a
+    /// partial write never leaves a corrupt state file. The <see cref="_stateWriteLock"/>
+    /// semaphore serialises concurrent flush calls so that only one write is
+    /// in-flight at a time.
     /// The file is written as JSON to <c>{Directory}/stream.state</c>.
     /// Reference: golang/nats-server/server/filestore.go:5820 (writeFullState).
     /// </summary>
-    private void WriteStreamState()
+    private async Task WriteStreamStateAsync()
     {
         var statePath = Path.Combine(_options.Directory, "stream.state");
-        var tmpPath = statePath + ".tmp";
 
         var snapshot = new StreamStateSnapshot
         {
@@ -1837,9 +1846,17 @@ public sealed class FileStore : IStreamStore, IAsyncDisposable, IDisposable
             Bytes = (ulong)_blocks.Sum(b => b.BytesUsed),
         };
 
-        var json = JsonSerializer.Serialize(snapshot);
-        File.WriteAllText(tmpPath, json);
-        File.Move(tmpPath, statePath, overwrite: true);
+        var json = JsonSerializer.SerializeToUtf8Bytes(snapshot);
+
+        await _stateWriteLock.WaitAsync();
+        try
+        {
+            await AtomicFileWriter.WriteAtomicallyAsync(statePath, json);
+        }
+        finally
+        {
+            _stateWriteLock.Release();
+        }
     }
 
     // -------------------------------------------------------------------------

src/NATS.Server/JetStream/Storage/IStreamStore.cs

@@ -56,7 +56,7 @@ public interface IStreamStore
         => throw new NotSupportedException("Block-engine SkipMsgs not yet implemented.");
 
     // Go: StreamStore.FlushAllPending — flush any buffered writes to backing storage
-    void FlushAllPending()
+    Task FlushAllPending()
         => throw new NotSupportedException("Block-engine FlushAllPending not yet implemented.");
 
     // Go: StreamStore.LoadMsg — load message by exact sequence; sm is an optional reusable buffer

src/NATS.Server/JetStream/Storage/MemStore.cs

@@ -355,7 +355,7 @@ public sealed class MemStore : IStreamStore
     }
 
     // Go: memStore.FlushAllPending server/memstore.go:423 — no-op for in-memory store
-    void IStreamStore.FlushAllPending() { }
+    Task IStreamStore.FlushAllPending() => Task.CompletedTask;
 
     // Go: memStore.LoadMsg server/memstore.go:1692
     StoreMsg IStreamStore.LoadMsg(ulong seq, StoreMsg? sm)