Gate cache hydration and preserve legacy generations

src/GameCult.Caching.MessagePack/DirectoryMessagePackBackingStore.cs

@@ -89,9 +89,7 @@ public sealed class DirectoryMessagePackBackingStore : CacheBackingStore
                     .OrderBy(record => record.Key, StringComparer.Ordinal)
                     .Select(record => ReadPersistedRecordPage(record, manifest.FormatVersion, out _))
                     .ToArray()
-                : manifest.Records
-                    .OrderBy(record => record.Key, StringComparer.Ordinal)
-                    .ToArray();
+                : ReadLegacyPersistedGenerationRecords(manifest);
             return new CultPersistedStoreSnapshot
             {
                 FormatVersion = manifest.FormatVersion,
@@ -101,6 +99,28 @@ public sealed class DirectoryMessagePackBackingStore : CacheBackingStore
         }
     }
 
+    private CultPersistedRecord[] ReadLegacyPersistedGenerationRecords(CultPersistedStoreSnapshot manifest)
+    {
+        var records = manifest.Records
+            .Where(record => !string.IsNullOrWhiteSpace(record.Key))
+            .ToList();
+        _recordDirectory.Refresh();
+        if (_recordDirectory.Exists)
+        {
+            records.AddRange(_recordDirectory.EnumerateFiles("*.msgpack")
+                .OrderBy(file => file.Name, StringComparer.Ordinal)
+                .Select(file => CultDocumentMessagePackSerialization.DeserializePersistedRecord(
+                    ReadAllBytesShared(file.FullName))));
+        }
+
+        return records
+            .Where(record => !string.IsNullOrWhiteSpace(record.Key))
+            .GroupBy(record => record.Key, StringComparer.Ordinal)
+            .Select(group => group.Last())
+            .OrderBy(record => record.Key, StringComparer.Ordinal)
+            .ToArray();
+    }
+
     /// <inheritdoc />
     public override void PullAll()
     {

src/GameCult.Caching/CultCache.cs

@@ -1446,13 +1446,22 @@ namespace GameCult.Caching
         /// </summary>
         public async Task PullAllBackingStoresAsync()
         {
-            foreach (var store in _backingStores)
-            {
-                store.PullAll();
-            }
+            if (_ambientTransaction.Value != null)
+                throw new InvalidOperationException(
+                    "CultCache hydration cannot run inside a mutation transaction; hydrate before opening the commit scope.");
 
-            _hasUnflushedMutations = _backingStores.Any(store => store.IsDirty);
-            await Task.CompletedTask;
+            await _transactionGate.WaitAsync().ConfigureAwait(false);
+            try
+            {
+                foreach (var store in _backingStores)
+                    store.PullAll();
+
+                _hasUnflushedMutations = _backingStores.Any(store => store.IsDirty);
+            }
+            finally
+            {
+                _transactionGate.Release();
+            }
         }
 
         /// <summary>
@@ -1462,13 +1471,22 @@ namespace GameCult.Caching
         public async Task PullBackingStoreRecordsAsync(Func<CultPersistedRecordMetadata, bool> selector)
         {
             if (selector == null) throw new ArgumentNullException(nameof(selector));
-            foreach (var store in _backingStores)
-            {
-                store.PullSelected(selector);
-            }
+            if (_ambientTransaction.Value != null)
+                throw new InvalidOperationException(
+                    "CultCache hydration cannot run inside a mutation transaction; hydrate before opening the commit scope.");
 
-            _hasUnflushedMutations = _backingStores.Any(store => store.IsDirty);
-            await Task.CompletedTask;
+            await _transactionGate.WaitAsync().ConfigureAwait(false);
+            try
+            {
+                foreach (var store in _backingStores)
+                    store.PullSelected(selector);
+
+                _hasUnflushedMutations = _backingStores.Any(store => store.IsDirty);
+            }
+            finally
+            {
+                _transactionGate.Release();
+            }
         }
 
         /// <summary>

tests/GameCult.Caching.Tests/BackingStoreTests.cs

@@ -574,6 +574,10 @@ namespace GameCult.Caching.Tests
                 File.WriteAllBytes(recordPath, CultDocumentMessagePackSerialization.SerializePersistedRecord(record));
 
                 var readStore = new DirectoryMessagePackBackingStore(filePath, recordsPath);
+                var rawGeneration = readStore.ReadPersistedGeneration();
+                Assert.That(rawGeneration.Records.Select(value => value.Key),
+                    Is.EqualTo(new[] { "record:stale" }),
+                    "v1 split record pages remain part of the selected legacy generation");
                 var readCache = new CultCache();
                 readCache.AddBackingStore(readStore);
                 await readCache.PullAllBackingStoresAsync();
@@ -1484,6 +1488,85 @@ namespace GameCult.Caching.Tests
             }
         }
 
+        [Test]
+        public async Task CultCache_Transaction_Rejects_Hydration_Inside_Its_Mutation_Scope()
+        {
+            var root = Path.Combine(Path.GetTempPath(), $"cultcache-transaction-hydration-{Guid.NewGuid():N}");
+            var manifest = Path.Combine(root, "state.cc");
+            Directory.CreateDirectory(root);
+            try
+            {
+                using (var writer = new CultCache())
+                {
+                    writer.AddBackingStore(new DirectoryMessagePackBackingStore(manifest));
+                    await writer.UpsertAsync(
+                        new NamedTestEntry { Name = "external", Value = "durable" },
+                        new CultRecordHandle<NamedTestEntry>(new CultRecordKey("transaction:external")));
+                    writer.FlushAllBackingStores();
+                }
+
+                using var reader = new CultCache();
+                reader.AddBackingStore(new DirectoryMessagePackBackingStore(manifest));
+                var observed = 0;
+                using var subscription = reader.Watch<NamedTestEntry>().Subscribe(_ => observed++);
+
+                Assert.That(
+                    async () => await reader.ExecuteTransactionAsync(
+                        () => reader.PullAllBackingStoresAsync()),
+                    Throws.TypeOf<InvalidOperationException>().With.Message.Contains("hydrate before"));
+                Assert.That(reader.Get<NamedTestEntry>(new CultRecordKey("transaction:external")), Is.Null);
+                Assert.That(observed, Is.Zero);
+
+                await reader.PullAllBackingStoresAsync();
+                Assert.That(reader.Get<NamedTestEntry>(new CultRecordKey("transaction:external"))?.Value,
+                    Is.EqualTo("durable"));
+                Assert.That(observed, Is.EqualTo(1));
+            }
+            finally
+            {
+                if (Directory.Exists(root)) Directory.Delete(root, recursive: true);
+            }
+        }
+
+        [Test]
+        public async Task CultCache_Hydration_Waits_For_The_Active_Mutation_Transaction()
+        {
+            var root = Path.Combine(Path.GetTempPath(), $"cultcache-hydration-gate-{Guid.NewGuid():N}");
+            var manifest = Path.Combine(root, "state.cc");
+            Directory.CreateDirectory(root);
+            try
+            {
+                using var cache = new CultCache();
+                cache.AddBackingStore(new DirectoryMessagePackBackingStore(manifest));
+                var staged = new TaskCompletionSource<bool>(TaskCreationOptions.RunContinuationsAsynchronously);
+                var release = new TaskCompletionSource<bool>(TaskCreationOptions.RunContinuationsAsynchronously);
+
+                var transaction = cache.ExecuteTransactionAsync(async () =>
+                {
+                    await cache.UpsertAsync(
+                        new NamedTestEntry { Name = "staged", Value = "committed" },
+                        new CultRecordHandle<NamedTestEntry>(new CultRecordKey("transaction:hydration-gate")));
+                    staged.SetResult(true);
+                    await release.Task;
+                });
+
+                await staged.Task;
+                var hydration = cache.PullAllBackingStoresAsync();
+                Assert.That(hydration.IsCompleted, Is.False,
+                    "hydration must wait until the active mutation transaction has committed");
+
+                release.SetResult(true);
+                await transaction;
+                await hydration;
+                Assert.That(cache.Get<NamedTestEntry>(new CultRecordKey("transaction:hydration-gate"))?.Value,
+                    Is.EqualTo("committed"));
+            }
+            finally
+            {
+                if (Directory.Exists(root)) Directory.Delete(root, recursive: true);
+            }
+        }
+
         [Test]
         public async Task CultCache_Transaction_Observer_Writes_In_A_New_Commit()
         {

unity/org.gamecult.cultlib/package.json

@@ -1,7 +1,7 @@
 {
   "name": "org.gamecult.cultlib",
   "displayName": "CultLib",
-  "version": "1.0.55",
+  "version": "1.0.56",
   "unity": "2021.3",
   "description": "CultCache, CultNet, and CultMesh managed runtime assemblies for Unity.",
   "license": "MIT",