GameCult/CultLib
Watch
9
0
Fork
You've already forked CultLib
 0
Code Issues Pull requests 1 Projects Releases Packages Wiki Activity Actions

Pin the single-owner publication cases Soul falsified

Browse source 
Observer rewrite of the door key, variant log order, replica log sequences with
and without a local write, 4x5000 concurrent writers, a write after a database
delete, shard-snapshot resync, and a disposed database.

Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>
This commit is contained in:
Meta Vens 2026-09-29 13:21:39 +02:00
commit d8d973c418
1 changed files with 199 additions and 0 deletions
Download patch file Download diff file Expand all files Collapse all files

199
tests/GameCult.Networking.Tests/CultNetPublishesCommitsTests.cs
View file

@ -10,6 +10,7 @@ using System.Threading.Tasks;
using GameCult.Caching;
using GameCult.Caching.MessagePack;
using NUnit.Framework;
using R3;
using static GameCult.Networking.Tests.NetworkingTests;
using static GameCult.Networking.Tests.CultNetDocumentRegistryVariantTests;
@ -226,5 +227,203 @@ namespace GameCult.Networking.Tests
Assert.That(Of(wire, baseKey.Value), Has.Length.EqualTo(1));
Assert.That(Of(wire, variantKey.Value), Is.Empty, "a variant is never carried as a record");
}
// An observer that rewrites the very key a door just wrote is a second write: it publishes and logs on its own.
[Test]
public async Task AnObserverRewriteOfTheDoorsOwnKeyIsPublishedAndLoggedToo()
{
var cache = new CultCache();
var database = new CultNetDatabase(cache);
using var wire = new Wire(database);
await wire.SubscribeAsync(false, typeof(NetworkSchemaNote));
var key = new CultRecordKey("tests:publish:normalized");
using var normalizer = cache.Watch<NetworkSchemaNote>().Subscribe(change =>
{
if (change.Document?.Text == "raw")
cache.UpsertAsync(Note("normalized"), new CultRecordHandle<NetworkSchemaNote>(change.Key)).GetAwaiter().GetResult();
});
await database.PutAsync(key, Note("raw"));
await Until(() => Of(wire, key.Value).Length >= 2);
await SettleAsync(database, wire);
Assert.That(cache.Get<NetworkSchemaNote>(key)!.Text, Is.EqualTo("normalized"));
Assert.That(Of(wire, key.Value).Select(change => change.ChangeKind), Is.EqualTo(new[] { "added", "updated" }));
Assert.That(database.GetMutationLog(database.Shards[0].ShardId).Count(entry => entry.Key.Value == key.Value), Is.EqualTo(2));
}
private static (CultCache Cache, CultNetDocumentRegistry Documents) VariantCache()
{
var registry = CultDocumentRegistry.Shared;
var cache = new CultCache(registry, CultCacheMessagePack.CreateCodec(registry));
var documents = new CultNetDocumentRegistry(registry)
.Register(CultNetDocumentBinding.ForDocument<VariantWireFixture>(registry));
return (cache, documents);
}
private static System.Collections.Generic.List<string> Record(CultNetDatabase database, Func<object, string> describe)
{
var seen = new System.Collections.Generic.List<string>();
database.WatchAllChanges().Subscribe(change => seen.Add(describe(change)));
return seen;
}
private static string KeyOfChange(object change) => ((dynamic)change).Key.Value;
[Test]
public async Task ABaseWriteAndItsReResolvedVariantAreLoggedInCacheOrder()
{
var (cache, documents) = VariantCache();
var database = new CultNetDatabase(cache, new CultNetDatabaseOptions { DocumentRegistry = documents });
var baseKey = new CultRecordKey("publish-variant-base");
var variantKey = new CultRecordKey("publish-variant-big");
await cache.UpsertAsync(new VariantWireFixture { Name = "base", Power = 1 }, new CultRecordHandle<VariantWireFixture>(baseKey));
await cache.UpsertVariantAsync(variantKey, baseKey, new[]
{
cache.Override<VariantWireFixture>(nameof(VariantWireFixture.Name), "big")
});
var published = Record(database, KeyOfChange);
var logged = database.GetMutationLog(database.Shards[0].ShardId).Count;
await database.PutAsync(baseKey, new VariantWireFixture { Name = "base edited", Power = 2 });
Assert.That(published, Is.EqualTo(new[] { baseKey.Value, variantKey.Value }));
Assert.That(database.GetMutationLog(database.Shards[0].ShardId, logged).Select(entry => entry.Key.Value),
Is.EqualTo(new[] { baseKey.Value, variantKey.Value }));
}
private static CultNetDatabase Replica(CultCache cache, CultNetDocumentRegistry? documents, string schemaId) =>
new(cache, new CultNetDatabaseOptions
{
DocumentRegistry = documents,
Shards =
[
new CultNetShardDescriptor("replica-shard", "runtime-a", epoch: 3, isPrimary: false, schemaIds: [schemaId])
]
});
[Test]
public async Task AReplicaLogHoldsOnlyThePrimarysSequencesWhateverItResolvesLocally()
{
var (sourceCache, sourceDocuments) = VariantCache();
var (replicaCache, replicaDocuments) = VariantCache();
var schemaId = sourceCache.Registry.GetRequired<VariantWireFixture>().SchemaId;
var source = new CultNetDatabase(sourceCache, new CultNetDatabaseOptions
{
DocumentRegistry = sourceDocuments,
Shards = [new CultNetShardDescriptor("replica-shard", "runtime-a", epoch: 3, isPrimary: true, schemaIds: [schemaId])]
});
var replica = Replica(replicaCache, replicaDocuments, schemaId);
var baseKey = new CultRecordKey("publish-variant-base");
var variantKey = new CultRecordKey("publish-variant-big");
await replicaCache.UpsertAsync(new VariantWireFixture { Name = "base", Power = 1 }, new CultRecordHandle<VariantWireFixture>(baseKey));
await replicaCache.UpsertVariantAsync(variantKey, baseKey, new[]
{
replicaCache.Override<VariantWireFixture>(nameof(VariantWireFixture.Name), "big")
});
await source.PutAsync(baseKey, new VariantWireFixture { Name = "base edited", Power = 2 });
await replica.ApplyShardLogResponseAsync(new CultNetShardLogResponseMessage
{
ShardId = "replica-shard",
ShardEpoch = 3,
Entries = source.GetMutationLogMessages("replica-shard").ToArray()
});
Assert.That(replicaCache.Get<VariantWireFixture>(variantKey)!.Power, Is.EqualTo(2), "the dependent re-resolved");
Assert.That(replica.GetMutationLog("replica-shard").Select(entry => entry.Sequence), Is.EqualTo(new[] { 1L }),
"the replica log holds the primary's entry and mints none of its own");
}
[Test]
public async Task ALocalCacheWriteOnAReplicaIsPublishedOnceAndNeverLogged()
{
var cache = new CultCache();
var schemaId = cache.Registry.GetRequired<NetworkSchemaNote>().SchemaId;
var replica = Replica(cache, null, schemaId);
var published = Record(replica, KeyOfChange);
await cache.UpsertAsync(Note("local"), new CultRecordHandle<NetworkSchemaNote>(new CultRecordKey("tests:publish:local")));
Assert.That(published, Is.EqualTo(new[] { "tests:publish:local" }));
Assert.That(replica.GetMutationLog("replica-shard"), Is.Empty);
}
[Test]
public void ConcurrentWritersLoseNoLogEntryAndReuseNoSequence()
{
var cache = new CultCache();
var database = new CultNetDatabase(cache);
const int writers = 4, perWriter = 5000;
Parallel.For(0, writers, writer =>
{
for (var i = 0; i < perWriter; i++)
cache.UpsertAsync(Note("n"), new CultRecordHandle<NetworkSchemaNote>(new CultRecordKey($"tests:publish:c:{writer}:{i}")))
.GetAwaiter().GetResult();
});
var sequences = database.GetMutationLog(database.Shards[0].ShardId).Select(entry => entry.Sequence).ToArray();
Assert.That(sequences, Has.Length.EqualTo(writers * perWriter));
Assert.That(sequences.Distinct().Count(), Is.EqualTo(writers * perWriter));
}
[Test]
public async Task AWriteAfterADatabaseDeleteIsStillPublishedAndLogged()
{
var cache = new CultCache();
var database = new CultNetDatabase(cache);
var key = new CultRecordKey("tests:publish:after-delete");
await database.PutAsync(key, Note("one"));
await database.DeleteAsync<NetworkSchemaNote>(key);
var published = Record(database, KeyOfChange);
await cache.UpsertAsync(Note("two"), new CultRecordHandle<NetworkSchemaNote>(key));
Assert.That(published, Is.EqualTo(new[] { key.Value }));
Assert.That(database.GetMutationLog(database.Shards[0].ShardId).Select(entry => entry.Kind),
Is.EqualTo(new[]
{
CultNetDatabaseChangeKind.Added, CultNetDatabaseChangeKind.Removed, CultNetDatabaseChangeKind.Added
}));
}
[Test]
public async Task AShardSnapshotResyncPublishesItsRemovalsAndPutsOnceAndLogsNothing()
{
var cache = new CultCache();
var schemaId = cache.Registry.GetRequired<NetworkSchemaNote>().SchemaId;
var replica = Replica(cache, null, schemaId);
var stale = new CultRecordKey("tests:publish:stale");
var fresh = new CultRecordKey("tests:publish:fresh");
await cache.UpsertAsync(Note("stale"), new CultRecordHandle<NetworkSchemaNote>(stale));
var published = Record(replica, change => ((dynamic)change).Kind.ToString() + ":" + ((dynamic)change).Key.Value);
var put = replica.Documents.CreateRawDocumentPutMessage(
"snapshot-fresh", new CultRecordHandle<NetworkSchemaNote>(fresh), Note("fresh"));
await replica.ApplyShardSnapshotResponseAsync(replica.Shards[0], new CultNetSnapshotResponseRawMessage
{
ShardId = "replica-shard",
ShardEpoch = 3,
Documents = [put.Document!],
ShardLogSequence = 5
});
Assert.That(published, Is.EqualTo(new[] { "Removed:" + stale.Value, "Added:" + fresh.Value }));
Assert.That(replica.GetMutationLog("replica-shard"), Is.Empty);
}
[Test]
public async Task ADisposedDatabaseNoLongerObservesTheCache()
{
var cache = new CultCache();
var database = new CultNetDatabase(cache);
var published = Record(database, KeyOfChange);
database.Dispose();
await cache.UpsertAsync(Note("after"), new CultRecordHandle<NetworkSchemaNote>(new CultRecordKey("tests:publish:disposed")));
Assert.That(published, Is.Empty);
}
}
}