fix(storage): don't let lock-free metadata readers miss records a compaction moves (follow-up to #196)

src/BLite.Core/Storage/StorageEngine.Collections.cs

@@ -134,6 +134,34 @@ public class IndexMetadata
 public sealed partial class StorageEngine
 {
     public CollectionMetadata? GetCollectionMetadata(string name)
+    {
+        // Lock-free first: saves only rewrite or move records, so a record this walk finds is the
+        // collection's. A miss proves nothing, though: a concurrent compaction (any save, under
+        // _metadataLock) can move the record onto a page the walk has already read and rewrite or
+        // free the page it was on. Callers take null for "new collection" and save fresh metadata
+        // over the existing record, so a miss, or a walk that runs into a page it cannot parse, is
+        // repeated under the lock.
+        try
+        {
+            var metadata = FindCollectionMetadata(name);
+            if (metadata != null)
+                return metadata;
+        }
+        catch
+        {
+            // Retried under the lock below, where an error is real and propagates.
+        }
+
+        if (!_metadataLock.Wait(_config.LockTimeout.WriteTimeoutMs))
+            throw new TimeoutException("Timed out acquiring metadata lock (GetCollectionMetadata).");
+        try
+        {
+            return FindCollectionMetadata(name);
+        }
+        finally { _metadataLock.Release(); }
+    }
+
+    private CollectionMetadata? FindCollectionMetadata(string name)
     {
         var buffer = new byte[PageSize];
         uint pageId = 1;
@@ -406,6 +434,7 @@ public void SaveCollectionMetadata(CollectionMetadata metadata)
         }
 
         var newData = stream.ToArray();
+        // Checked before anything is written or allocated: the stored record stays as it is.
         if (newData.Length + SlotEntry.Size > PageSize - SlottedPageHeader.Size)
             throw new InvalidOperationException(
                 $"Collection metadata for '{metadata.Name}' ({newData.Length} bytes) does not fit in a page.");
@@ -539,16 +568,26 @@ private static (int Page, int Slot) FindMetadataRecord(List<(uint PageId, byte[]
     /// Pages are rewritten front to back and every record lands on its own page or an earlier one,
     /// so all records are on some page after each write. A crash part-way can leave a record on
     /// two pages; the copies are identical, readers take the first, and the next compaction drops
-    /// the other.
+    /// the other. A crash after the last page is rewritten but before the pages past the new end
+    /// are freed leaves those pages unreachable and off the free list: they leak (file space only,
+    /// no records are lost), which is accepted to keep the rewrite free of extra bookkeeping.
     /// </remarks>
     private List<(uint PageId, byte[] Buffer)> CompactMetadataChain(List<(uint PageId, byte[] Buffer)> chain, bool onlyIfShorter)
     {
-        var records = new List<byte[]>();
-        var names   = new HashSet<string>(StringComparer.OrdinalIgnoreCase);
+        // A single page cannot get shorter. This is the common case: every in-place save asks.
+        if (onlyIfShorter && chain.Count <= 1)
+            return chain;
+
+        // Plan from the slot tables alone: which records stay, and the page each one lands on.
+        var keep  = new List<(int SourcePage, SlotEntry Slot, int TargetPage)>();
+        var names = new HashSet<string>(StringComparer.OrdinalIgnoreCase);
         bool anythingToDrop = false;
+        int pageCount = 1;
+        int room = PageSize - SlottedPageHeader.Size;
 
-        foreach (var (_, buffer) in chain)
+        for (int p = 0; p < chain.Count; p++)
         {
+            var buffer = chain[p].Buffer;
             var header = SlottedPageHeader.ReadFrom(buffer);
             int liveBytes = 0;
             for (int i = 0; i < header.SlotCount; i++)
@@ -558,46 +597,45 @@ private static (int Page, int Slot) FindMetadataRecord(List<(uint PageId, byte[]
 
                 var name = TryReadRecordName(buffer, slot);
                 if (name != null && !names.Add(name)) { anythingToDrop = true; continue; }
-
-                records.Add(buffer.AsSpan(slot.Offset, slot.Length).ToArray());
                 liveBytes += slot.Length;
-            }
-            if (PageSize - header.FreeSpaceEnd > liveBytes) anythingToDrop = true;
-        }
 
-        // Greedy packing in order needs no more pages than any in-order layout, the current one included.
-        var pages = new List<List<byte[]>> { new() };
-        int room = PageSize - SlottedPageHeader.Size;
-        foreach (var record in records)
-        {
-            int need = record.Length + SlotEntry.Size;
-            if (need > room)
-            {
-                pages.Add(new List<byte[]>());
-                room = PageSize - SlottedPageHeader.Size;
+                // Greedy packing in order needs no more pages than any in-order layout, the current one included.
+                int need = slot.Length + SlotEntry.Size;
+                if (need > room)
+                {
+                    pageCount++;
+                    room = PageSize - SlottedPageHeader.Size;
+                }
+                room -= need;
+                keep.Add((p, slot, pageCount - 1));
             }
-            pages[pages.Count - 1].Add(record);
-            room -= need;
+            if (PageSize - header.FreeSpaceEnd > liveBytes) anythingToDrop = true;
         }
 
-        if (pages.Count >= chain.Count && (onlyIfShorter || !anythingToDrop))
+        if (pageCount >= chain.Count && (onlyIfShorter || !anythingToDrop))
             return chain;
 
-        var result = new List<(uint PageId, byte[] Buffer)>(pages.Count);
-        for (int p = 0; p < pages.Count; p++)
+        // The records are copied out of the buffers loaded before the first write; each page is
+        // rebuilt in a fresh buffer, so those stay intact while the chain is rewritten.
+        var result = new List<(uint PageId, byte[] Buffer)>(pageCount);
+        int k = 0;
+        for (int p = 0; p < pageCount; p++)
         {
             var pageId = chain[p].PageId;
-            var next   = p + 1 < pages.Count ? chain[p + 1].PageId : 0u;
+            var next   = p + 1 < pageCount ? chain[p + 1].PageId : 0u;
             var buffer = new byte[PageSize];
             NewMetadataPage(pageId, next).WriteTo(buffer);
-            foreach (var record in pages[p])
-                AppendRecord(buffer, record);
+            for (; k < keep.Count && keep[k].TargetPage == p; k++)
+            {
+                var (source, slot, _) = keep[k];
+                AppendRecord(buffer, chain[source].Buffer.AsSpan(slot.Offset, slot.Length));
+            }
             WritePageImmediate(pageId, buffer);
             result.Add((pageId, buffer));
         }
 
         // Page 1 is always kept; the pages after the new end of the chain are no longer reachable.
-        for (int p = pages.Count; p < chain.Count; p++)
+        for (int p = pageCount; p < chain.Count; p++)
             FreePage(chain[p].PageId);
 
         return result;
@@ -615,7 +653,7 @@ private static (int Page, int Slot) FindMetadataRecord(List<(uint PageId, byte[]
     };
 
     /// <summary>Adds <paramref name="record"/> as a new slot; the caller checked that it fits.</summary>
-    private static void AppendRecord(byte[] page, byte[] record)
+    private static void AppendRecord(byte[] page, ReadOnlySpan<byte> record)
     {
         var header = SlottedPageHeader.ReadFrom(page);
         int offset = header.FreeSpaceEnd - record.Length;
@@ -754,6 +792,12 @@ public void DeleteCollectionMetadata(string name)
 
     public IReadOnlyList<CollectionMetadata> GetAllCollectionsMetadata()
     {
+        // Under the lock: a concurrent compaction can move records onto a page this walk has
+        // already read, and unlike a single lookup the result would not show that they are missing.
+        if (!_metadataLock.Wait(_config.LockTimeout.WriteTimeoutMs))
+            throw new TimeoutException("Timed out acquiring metadata lock (GetAllCollectionsMetadata).");
+        try
+        {
         var result = new List<CollectionMetadata>();
         // One entry per collection: the first copy, as GetCollectionMetadata returns it (a crash
         // while the chain was compacted can leave a second one; see CompactMetadataChain).
@@ -795,6 +839,8 @@ public IReadOnlyList<CollectionMetadata> GetAllCollectionsMetadata()
         }
 
         return result;
+        } // end try
+        finally { _metadataLock.Release(); }
     }
 
     /// <summary>

tests/BLite.Tests/MetadataChainTests.cs

@@ -1,6 +1,7 @@
 using BLite.Core;
 using BLite.Core.Retention;
 using BLite.Core.Storage;
+using BLite.Core.Transactions;
 using BLite.Shared;
 
 namespace BLite.Tests;
@@ -360,4 +361,213 @@ public async Task DuplicateRecord_IsListedOnce_DeletedWithTheOriginal_AndDropped
         Assert.Null(storage.GetCollectionMetadata("c1"));
         Assert.Equal(new[] { "c0", "c2" }, storage.GetAllCollectionsMetadata().Select(m => m.Name).OrderBy(n => n));
     }
+
+    // ── Readers that do not take the metadata lock ───────────────────────────
+
+    /// <summary>In-memory page storage that runs a callback once, right after the next read of a page.</summary>
+    private sealed class InterceptingPageStorage : IPageStorage
+    {
+        private readonly MemoryPageStorage _inner;
+        private uint _pageId;
+        private Action? _afterRead;
+
+        public InterceptingPageStorage(int pageSize)
+        {
+            _inner = new MemoryPageStorage(pageSize);
+            _inner.Open();
+        }
+
+        public void AfterNextReadOf(uint pageId, Action action)
+        {
+            _pageId = pageId;
+            Volatile.Write(ref _afterRead, action);
+        }
+
+        public void ReadPage(uint pageId, Span<byte> destination)
+        {
+            _inner.ReadPage(pageId, destination);
+            if (pageId == _pageId && Interlocked.Exchange(ref _afterRead, null) is { } action)
+                action();
+        }
+
+        public int PageSize => _inner.PageSize;
+        public uint NextPageId => _inner.NextPageId;
+        public void Open() { }
+        public void ReadPageHeader(uint pageId, Span<byte> destination) => _inner.ReadPageHeader(pageId, destination);
+        public ValueTask ReadPageAsync(uint pageId, Memory<byte> destination, CancellationToken cancellationToken = default)
+            => _inner.ReadPageAsync(pageId, destination, cancellationToken);
+        public void WritePage(uint pageId, ReadOnlySpan<byte> source) => _inner.WritePage(pageId, source);
+        public uint AllocatePage() => _inner.AllocatePage();
+        public void FreePage(uint pageId) => _inner.FreePage(pageId);
+        public void Flush() => _inner.Flush();
+        public Task FlushAsync(CancellationToken cancellationToken = default) => _inner.FlushAsync(cancellationToken);
+        public Task BackupAsync(string destinationPath, CancellationToken cancellationToken = default)
+            => _inner.BackupAsync(destinationPath, cancellationToken);
+        public Task TruncateToMinimumAsync(CancellationToken cancellationToken = default)
+            => _inner.TruncateToMinimumAsync(cancellationToken);
+        public void Dispose() => _inner.Dispose();
+    }
+
+    /// <summary>The live collection names on each page of the chain.</summary>
+    private static List<(uint PageId, List<string> Names)> Layout(StorageEngine storage)
+    {
+        var layout = new List<(uint, List<string>)>();
+        var buffer = new byte[storage.PageSize];
+        uint pageId = 1;
+        while (pageId != 0)
+        {
+            storage.ReadPage(pageId, null, buffer);
+            var header = SlottedPageHeader.ReadFrom(buffer);
+            if (header.PageType != PageType.Collection) break;
+            var names = new List<string>();
+            for (int i = 0; i < header.SlotCount; i++)
+            {
+                var slot = SlotEntry.ReadFrom(buffer.AsSpan(SlottedPageHeader.Size + i * SlotEntry.Size));
+                if ((slot.Flags & SlotFlags.Deleted) != 0) continue;
+                using var reader = new BinaryReader(new MemoryStream(buffer, slot.Offset, slot.Length, false));
+                names.Add(reader.ReadString());
+            }
+            layout.Add((pageId, names));
+            pageId = header.NextOverflowPage;
+        }
+        return layout;
+    }
+
+    /// <summary>
+    /// A chain of three pages whose live records fit in two: all of page 1 but its first record is
+    /// deleted. The next save of that first record compacts the chain: the records of page 2 move
+    /// to page 1 (but its last), page 3's move to page 2, and page 3 is freed.
+    /// </summary>
+    private static (StorageEngine Storage, InterceptingPageStorage Pages, List<(uint PageId, List<string> Names)> Layout)
+        ChainAboutToBeCompacted()
+    {
+        var pages = new InterceptingPageStorage(16384);
+        var storage = new StorageEngine(pages, new MemoryWriteAheadLog());
+        foreach (var name in Names(36))
+        {
+            storage.SaveCollectionMetadata(new CollectionMetadata
+            {
+                Name = name,
+                GeneralRetentionPolicy = new RetentionPolicy { TimestampField = new string('x', 1100) }
+            });
+        }
+
+        var layout = Layout(storage);
+        Assert.Equal(3, layout.Count);
+        foreach (var name in layout[0].Names.Skip(1))
+            storage.DeleteCollectionMetadata(name);
+        return (storage, pages, Layout(storage));
+    }
+
+    [Theory]
+    [InlineData(0)]   // the reader then reads page 2 after the rewrite: the record has left it for page 1
+    [InlineData(1)]   // the reader then follows the old link to page 3, which the compaction freed
+    public void GetCollectionMetadata_FindsARecordThatACompactionMovesBehindTheReader(int compactAfterPage)
+    {
+        var (storage, pages, layout) = ChainAboutToBeCompacted();
+        using var _ = storage;
+        var trigger = layout[0].Names[0];
+        var moved   = layout[compactAfterPage + 1].Names[0];   // moves onto the page the reader has just read
+
+        // A save of another collection compacts the chain while the reader is between two pages.
+        pages.AfterNextReadOf(layout[compactAfterPage].PageId,
+            () => Assert.True(Task.Run(() => storage.GetNextSequence(trigger)).Wait(TimeSpan.FromSeconds(10))));
+        var metadata = storage.GetCollectionMetadata(moved);
+
+        Assert.Equal(2, Layout(storage).Count);   // the compaction ran mid-walk
+        Assert.NotNull(metadata);
+        Assert.Equal(moved, metadata!.Name);
+        Assert.Equal(new string('x', 1100), metadata.GeneralRetentionPolicy!.TimestampField);
+    }
+
+    [Fact]
+    public void GetAllCollectionsMetadata_ListsRecordsThatACompactionMovesBehindTheReader()
+    {
+        var (storage, pages, layout) = ChainAboutToBeCompacted();
+        using var _ = storage;
+        var live = layout.SelectMany(p => p.Names).OrderBy(n => n).ToList();
+
+        // Same race as above, while the reader is between page 1 and page 2. The reader holds the
+        // metadata lock, so the save waits for it (well within the lock timeout) instead of running.
+        Task? compaction = null;
+        pages.AfterNextReadOf(1, () =>
+        {
+            compaction = Task.Run(() => storage.GetNextSequence(layout[0].Names[0]));
+            compaction.Wait(TimeSpan.FromMilliseconds(100));
+        });
+        var all = storage.GetAllCollectionsMetadata();
+        Assert.NotNull(compaction);
+        compaction!.Wait();
+
+        Assert.Equal(live, all.Select(m => m.Name).OrderBy(n => n));
+        Assert.Equal(2, Layout(storage).Count);
+    }
+
+    [Fact]
+    public async Task DeleteOfAMissingCollection_DoesNotWaitForTheLockItHolds()
+    {
+        // DeleteCollectionMetadata looks the record up while holding the metadata lock; a lookup
+        // that misses must not try to take that lock again (it is not reentrant).
+        await SeedAsync(Names(2));
+        using var storage = new StorageEngine(_dbPath, PageFileConfig.Default);
+
+        var watch = System.Diagnostics.Stopwatch.StartNew();
+        storage.DeleteCollectionMetadata("missing");
+        Assert.True(watch.ElapsedMilliseconds < 400, $"took {watch.ElapsedMilliseconds} ms");
+        Assert.Equal(new[] { "c0", "c1" }, storage.GetAllCollectionsMetadata().Select(m => m.Name).OrderBy(n => n));
+    }
+
+    // ── Records too large for a page ─────────────────────────────────────────
+
+    private static int RecordLength(StorageEngine storage, string collection)
+    {
+        var buffer = new byte[storage.PageSize];
+        uint pageId = 1;
+        while (pageId != 0)
+        {
+            storage.ReadPage(pageId, null, buffer);
+            var header = SlottedPageHeader.ReadFrom(buffer);
+            for (int i = 0; i < header.SlotCount; i++)
+            {
+                var slot = SlotEntry.ReadFrom(buffer.AsSpan(SlottedPageHeader.Size + i * SlotEntry.Size));
+                if ((slot.Flags & SlotFlags.Deleted) != 0) continue;
+                using var reader = new BinaryReader(new MemoryStream(buffer, slot.Offset, slot.Length, false));
+                if (reader.ReadString() == collection) return slot.Length;
+            }
+            pageId = header.NextOverflowPage;
+        }
+        throw new InvalidOperationException($"No record for {collection}.");
+    }
+
+    [Fact]
+    public async Task RecordTooLargeForAPage_IsRefusedBeforeAnythingIsWritten()
+    {
+        await SeedAsync(Names(3));
+        using var storage = new StorageEngine(_dbPath, PageFileConfig.Default);
+        int max = storage.PageSize - SlottedPageHeader.Size - SlotEntry.Size;
+
+        var meta = storage.GetCollectionMetadata("c1")!;
+        meta.GeneralRetentionPolicy = new RetentionPolicy { TimestampField = new string('t', 200) };
+        storage.SaveCollectionMetadata(meta);
+        int fill = 200 + max - RecordLength(storage, "c1");   // 2-byte length prefix either way
+
+        // The largest record a page holds is stored (on a page of its own)...
+        meta.GeneralRetentionPolicy.TimestampField = new string('t', fill);
+        storage.SaveCollectionMetadata(meta);
+        Assert.Equal(max, RecordLength(storage, "c1"));
+        var before = Inspect(storage);
+        Assert.Equal(2, before.Pages.Count);
+
+        // ...one byte more is refused, and the chain and the stored record stay exactly as they were.
+        meta.GeneralRetentionPolicy.TimestampField = new string('t', fill + 1);
+        var error = Assert.Throws<InvalidOperationException>(() => storage.SaveCollectionMetadata(meta));
+        Assert.Contains("does not fit in a page", error.Message);
+
+        var after = Inspect(storage);
+        Assert.Equal(before.Pages, after.Pages);
+        Assert.Equal(before.Live, after.Live);
+        Assert.Equal(before.Deleted, after.Deleted);
+        Assert.Equal(new string('t', fill), storage.GetCollectionMetadata("c1")!.GeneralRetentionPolicy!.TimestampField);
+        Assert.Equal(3, storage.GetAllCollectionsMetadata().Count);
+    }
 }