diff --git a/src/BLite.Core/Storage/StorageEngine.Collections.cs b/src/BLite.Core/Storage/StorageEngine.Collections.cs index 15354dd..26c21d0 100644 --- a/src/BLite.Core/Storage/StorageEngine.Collections.cs +++ b/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. /// private List<(uint PageId, byte[] Buffer)> CompactMetadataChain(List<(uint PageId, byte[] Buffer)> chain, bool onlyIfShorter) { - var records = new List(); - var names = new HashSet(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(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> { new() }; - int room = PageSize - SlottedPageHeader.Size; - foreach (var record in records) - { - int need = record.Length + SlotEntry.Size; - if (need > room) - { - pages.Add(new List()); - 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[] }; /// Adds as a new slot; the caller checked that it fits. - private static void AppendRecord(byte[] page, byte[] record) + private static void AppendRecord(byte[] page, ReadOnlySpan record) { var header = SlottedPageHeader.ReadFrom(page); int offset = header.FreeSpaceEnd - record.Length; @@ -754,6 +792,12 @@ public void DeleteCollectionMetadata(string name) public IReadOnlyList 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(); // 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 GetAllCollectionsMetadata() } return result; + } // end try + finally { _metadataLock.Release(); } } /// diff --git a/tests/BLite.Tests/MetadataChainTests.cs b/tests/BLite.Tests/MetadataChainTests.cs index d4f56da..222126e 100644 --- a/tests/BLite.Tests/MetadataChainTests.cs +++ b/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 ─────────────────────────── + + /// In-memory page storage that runs a callback once, right after the next read of a page. + 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 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 destination) => _inner.ReadPageHeader(pageId, destination); + public ValueTask ReadPageAsync(uint pageId, Memory destination, CancellationToken cancellationToken = default) + => _inner.ReadPageAsync(pageId, destination, cancellationToken); + public void WritePage(uint pageId, ReadOnlySpan 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(); + } + + /// The live collection names on each page of the chain. + private static List<(uint PageId, List Names)> Layout(StorageEngine storage) + { + var layout = new List<(uint, List)>(); + 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(); + 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; + } + + /// + /// 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. + /// + private static (StorageEngine Storage, InterceptingPageStorage Pages, List<(uint PageId, List 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(() => 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); + } }