Strengthen NTFS journal replay flush propagation and retry coverage

Library/DiscUtils.Ntfs/NtfsJournal.cs

@@ -128,7 +128,7 @@ internal NtfsJournalInfo(NtfsJournalStatus status, string detail)
     public IReadOnlyList<long> FileRecordsChanged { get; internal set; } = [];
 
     /// <summary>
-    /// Gets a value indicating whether replay wrote to the volume, reset the journal and completed its flush calls.
+    /// Gets a value indicating whether replay wrote to the volume stream, reset the journal and completed its flush calls.
     /// This is also true for a reset with no metadata changes (<see cref="NtfsJournalStatus.NothingToReplay"/>).
     /// </summary>
     public bool Replayed { get; internal set; }
@@ -177,8 +177,10 @@ public static class NtfsJournal
     /// Replay prepares the changes before writing. If the MFT mirror changes, it is written and flushed before
     /// the primary records. Replay flushes metadata before resetting the restart pages. It writes and flushes
     /// the alternate restart page before replacing and flushing the source page.
-    /// A <see cref="FileStream"/> uses <c>Flush(true)</c>. For other streams, durability and ordering after a crash
-    /// depend on whether <see cref="Stream.Flush()"/> reaches all backing storage layers.
+    /// Replay requests <see cref="SparseStream.Flush(bool)"/> with <c>true</c> through a non-owning sparse
+    /// stream adapter. Built-in wrappers forward the request to backing streams that support disk flushing;
+    /// other streams retain their parameterless <see cref="Stream.Flush()"/> behavior. The caller must establish
+    /// whether every storage layer honors the barrier.
     /// Replay is not atomic. An I/O error during the writes or flushes can leave a partially updated volume.
     /// Journals that require metadata undo or fail the local hibernation check cannot be written by this API.
     /// </remarks>
@@ -355,27 +357,31 @@ private static NtfsJournalInfo Run(Stream volume, bool write, out Dictionary<lon
         changedClusters = clusters;
         if (write)
         {
+            // This adapter owns no resources. Do not dispose it: a non-owning SparseWrapperStream
+            // performs an extra parameterless flush on disposal, which would change replay ordering
+            // and could hide the original exception from a failed durability barrier.
+            var flushStream = SparseStream.FromStream(volume, Ownership.None);
             bool IsMirrorCluster(KeyValuePair<long, byte[]> cluster) => mirrorClusterCount > 0 &&
                 cluster.Key >= layout.MftMirrorCluster && cluster.Key - layout.MftMirrorCluster < mirrorClusterCount;
 
             if (mirrorClusterCount > 0)
             {
                 // A retry can skip primary records whose LSN already advanced. Their copies must be written first.
                 WriteClusters(volume, clusters.Where(IsMirrorCluster), clusterSize);
-                FlushReplayWrites(volume);
+                flushStream.Flush(flushToDisk: true);
             }
 
             WriteClusters(volume, clusters.Where(cluster => !IsMirrorCluster(cluster)), clusterSize);
-            // Flush even when no pages changed. A previous attempt may have left visible but buffered metadata.
-            FlushReplayWrites(volume);
+            // Flush even without page changes. Metadata or a recognized alternate reset may still be buffered.
+            flushStream.Flush(flushToDisk: true);
             if (alternateRestart != null)
             {
                 storage.Write(alternateOffset, alternateRestart, 0, alternateRestart.Length);
-                FlushReplayWrites(volume);
+                flushStream.Flush(flushToDisk: true);
             }
 
             storage.Write(sourceOffset, sourceRestart, 0, sourceRestart.Length);
-            FlushReplayWrites(volume);
+            flushStream.Flush(flushToDisk: true);
             info.Replayed = true;
         }
 
@@ -407,18 +413,6 @@ private static SnapshotStream CreateReplayOverlay(Stream volume, Dictionary<long
         return snapshot;
     }
 
-    private static void FlushReplayWrites(Stream volume)
-    {
-        if (volume is FileStream file)
-        {
-            file.Flush(flushToDisk: true);
-        }
-        else
-        {
-            volume.Flush();
-        }
-    }
-
     private static void WriteClusters(Stream stream, IEnumerable<KeyValuePair<long, byte[]>> clusters, int clusterSize)
     {
         foreach (var cluster in clusters.OrderBy(c => c.Key))
@@ -433,9 +427,12 @@ private static void WriteClusters(Stream stream, IEnumerable<KeyValuePair<long,
     /// </summary>
     private static long AddMftMirror(Stream volume, JournalVolumeLayout layout, Dictionary<long, byte[]> clusters)
     {
-        var mirrorBytes = 4 * layout.MftRecordSize;
+        const int reservedRecords = 16;
+        var minimumMirrorBytes = 4 * layout.MftRecordSize;
         var clusterSize = layout.BytesPerCluster;
-        var mirrorClusters = (mirrorBytes + (long)clusterSize - 1) / clusterSize;
+        // A large cluster can mirror additional reserved records. Leave the remaining cluster bytes as stored.
+        var reservedMirrorBytes = Math.Max(minimumMirrorBytes, Math.Min(clusterSize, reservedRecords * layout.MftRecordSize));
+        var mirrorClusters = (reservedMirrorBytes + (long)clusterSize - 1) / clusterSize;
         var touched = clusters.Keys.Any(lcn => lcn >= layout.MftCluster && lcn - layout.MftCluster < mirrorClusters);
         if (!touched)
         {
@@ -450,13 +447,13 @@ private static long AddMftMirror(Stream volume, JournalVolumeLayout layout, Dict
         }
 
         var mftStart = layout.MftCluster * clusterSize;
-        var bytes = new byte[mirrorBytes];
-        for (var done = 0; done < mirrorBytes;)
+        var bytes = new byte[reservedMirrorBytes];
+        for (var done = 0; done < bytes.Length;)
         {
             var position = mftStart + done;
             var lcn = position / clusterSize;
             var inCluster = (int)(position % clusterSize);
-            var chunk = Math.Min(mirrorBytes - done, clusterSize - inCluster);
+            var chunk = Math.Min(bytes.Length - done, clusterSize - inCluster);
             if (clusters.TryGetValue(lcn, out var cluster))
             {
                 Array.Copy(cluster, inCluster, bytes, done, chunk);
@@ -478,11 +475,12 @@ private static long AddMftMirror(Stream volume, JournalVolumeLayout layout, Dict
         }
 
         var (extents, length) = LogFileStorage.ReadDataAttribute(mirrorRecord, layout, "$MFTMirr");
-        if (length < mirrorBytes || extents[0].Lcn != layout.MftMirrorCluster || extents[0].Length < mirrorClusters)
+        if (length < minimumMirrorBytes || extents[0].Lcn != layout.MftMirrorCluster || extents[0].Length < mirrorClusters)
         {
             throw new JournalException("$MFTMirr data does not match the contiguous mirror location in the boot sector");
         }
 
+        var mirrorBytes = (int)Math.Min(reservedMirrorBytes, length - length % layout.MftRecordSize);
         var mirrorStart = layout.MftMirrorCluster * clusterSize;
         for (var recordOffset = 0; recordOffset < mirrorBytes; recordOffset += mirrorRecord.Length)
         {

Library/DiscUtils.Streams/Block/BlockCacheStream.cs

@@ -571,6 +571,16 @@ public override void Flush()
         }
     }
 
+    public override void Flush(bool flushToDisk)
+    {
+        CheckDisposed();
+
+        if (CanWrite)
+        {
+            _wrappedStream.Flush(flushToDisk);
+        }
+    }
+
     /// <summary>
     /// Moves the stream position.
     /// </summary>
@@ -814,4 +824,4 @@ private void InvalidateBlocks(long firstBlock, int numBlocks)
             _cache.ReleaseBlock(i);
         }
     }
-}
+}

Library/DiscUtils.Streams/ConcatStream.cs

@@ -135,6 +135,15 @@ public override void Flush()
         }
     }
 
+    public override void Flush(bool flushToDisk)
+    {
+        CheckDisposed();
+        for (var i = 0; i < _streams.Count; ++i)
+        {
+            _streams[i].Flush(flushToDisk);
+        }
+    }
+
     public override int Read(byte[] buffer, int offset, int count)
     {
         CheckDisposed();
@@ -396,4 +405,4 @@ private void CheckDisposed()
         }
 #endif
     }
-}
+}

Library/DiscUtils.Streams/MirrorStream.cs

@@ -98,6 +98,14 @@ public override void Flush()
         }
     }
 
+    public override void Flush(bool flushToDisk)
+    {
+        foreach (var stream in _wrapped)
+        {
+            stream.Flush(flushToDisk);
+        }
+    }
+
     public override int Read(byte[] buffer, int offset, int count)
     {
         return _wrapped[0].Read(buffer, offset, count);
@@ -240,4 +248,4 @@ protected override void Dispose(bool disposing)
             base.Dispose(disposing);
         }
     }
-}
+}

Library/DiscUtils.Streams/SnapshotStream.cs

@@ -253,6 +253,13 @@ public override void Flush()
         _baseStream.Flush();
     }
 
+    public override void Flush(bool flushToDisk)
+    {
+        CheckFrozen();
+
+        FlushWrappedStream(_baseStream, flushToDisk);
+    }
+
     /// <summary>
     /// Flushes the stream.
     /// </summary>
@@ -622,4 +629,4 @@ private void CheckFrozen()
             throw new InvalidOperationException("The stream is frozen");
         }
     }
-}
+}

Library/DiscUtils.Streams/SparseStream.cs

@@ -48,6 +48,33 @@
     /// <remarks>This may be an empty enumeration if all bytes are zero.</remarks>
     public abstract IEnumerable<StreamExtent> Extents { get; }
 
+    /// <summary>
+    /// Flushes pending changes, requesting synchronization with backing storage when supported.
+    /// </summary>
+    /// <param name="flushToDisk">Whether to request a disk flush from supporting backing streams.</param>
+    /// <remarks>
+    /// The default implementation calls <see cref="Flush()"/>. A stream without a stronger flush
+    /// contract cannot guarantee that its data has reached durable storage.
+    /// </remarks>
+    public virtual void Flush(bool flushToDisk) => Flush();
+
+    /// <summary>Forwards a flush without bypassing intervening sparse-stream wrappers.</summary>
+    protected static void FlushWrappedStream(Stream wrapped, bool flushToDisk)
+    {
+        if (wrapped is SparseStream sparse)
+        {
+            sparse.Flush(flushToDisk);
+        }
+        else if (wrapped is FileStream file)
+        {
+            file.Flush(flushToDisk);
+        }
+        else
+        {
+            wrapped.Flush();
+        }
+    }
+
     /// <summary>
     /// Converts any stream into a sparse stream.
     /// </summary>
@@ -483,6 +510,16 @@
             }
         }
 
+        public override void Flush(bool flushToDisk)
+        {
+            CheckDisposed();
+
+            if (CanWrite)
+            {
+                FlushWrappedStream(_wrapped, flushToDisk);
+            }
+        }
+
         public override int Read(byte[] buffer, int offset, int count)
         {
             CheckDisposed();
@@ -688,6 +725,20 @@
             }
         }
 
+        public override void Flush(bool flushToDisk)
+        {
+            _sync.Wait();
+
+            try
+            {
+                content.Flush(flushToDisk);
+            }
+            finally
+            {
+                _sync.Release();
+            }
+        }
+
         public override async Task FlushAsync(CancellationToken cancellationToken)
         {
             await _sync.WaitAsync(cancellationToken).ConfigureAwait(false);
@@ -1043,4 +1094,4 @@
 
         public override string ToString() => $"Syncrhonized[{content}]";
     }
-}
+}

Library/DiscUtils.Streams/StripedStream.cs

@@ -116,6 +116,14 @@ public override void Flush()
         }
     }
 
+    public override void Flush(bool flushToDisk)
+    {
+        foreach (var stream in _wrapped)
+        {
+            stream.Flush(flushToDisk);
+        }
+    }
+
     public override int Read(byte[] buffer, int offset, int count)
     {
         if (!CanRead)
@@ -357,4 +365,4 @@ protected override void Dispose(bool disposing)
             base.Dispose(disposing);
         }
     }
-}
+}

Library/DiscUtils.Streams/SubStream.cs

@@ -109,6 +109,11 @@ public override void Flush()
         Parent.Flush();
     }
 
+    public override void Flush(bool flushToDisk)
+    {
+        FlushWrappedStream(Parent, flushToDisk);
+    }
+
     public override int Read(byte[] buffer, int offset, int count)
     {
         if (count < 0)
@@ -285,4 +290,4 @@ private IEnumerable<StreamExtent> OffsetExtents(IEnumerable<StreamExtent> src)
             yield return new StreamExtent(e.Start - _first, e.Length);
         }
     }
-}
+}

Library/DiscUtils.Streams/ThreadSafeStream.cs

@@ -239,6 +239,14 @@ public override void Flush()
         }
     }
 
+    public override void Flush(bool flushToDisk)
+    {
+        lock (_common)
+        {
+            Wrapped.Flush(flushToDisk);
+        }
+    }
+
     /// <summary>
     /// Reads data from the stream.
     /// </summary>
@@ -398,4 +406,4 @@ public void Dispose()
 
         #endregion
     }
-}
+}

Library/DiscUtils.Streams/WrappingMappedStream.cs

@@ -114,6 +114,11 @@ public override void Flush()
         WrappedStream.Flush();
     }
 
+    public override void Flush(bool flushToDisk)
+    {
+        FlushWrappedStream(WrappedStream, flushToDisk);
+    }
+
     public override Task FlushAsync(CancellationToken cancellationToken) =>
         WrappedStream.FlushAsync(cancellationToken);
 
@@ -205,4 +210,4 @@ protected override void Dispose(bool disposing)
             base.Dispose(disposing);
         }
     }
-}
+}

Library/DiscUtils.Streams/WrappingStream.cs

@@ -77,6 +77,11 @@ public override void Flush()
         _wrapped.Flush();
     }
 
+    public override void Flush(bool flushToDisk)
+    {
+        _wrapped.Flush(flushToDisk);
+    }
+
     public override int Read(byte[] buffer, int offset, int count)
     {
         return _wrapped.Read(buffer, offset, count);
@@ -161,4 +166,4 @@ protected override void Dispose(bool disposing)
             base.Dispose(disposing);
         }
     }
-}
+}