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

Add CultNet database subscriptions

Browse source
This commit is contained in:
Meta Vens 2026-05-19 17:10:24 +01:00
commit b54ed59135
7 changed files with 356 additions and 12 deletions
Download patch file Download diff file Expand all files Collapse all files

20
src/GameCult.Networking/Contracts/cultnet-distributed-database.md
View file

@ -37,10 +37,11 @@ Implemented:
- `CultNetDatabaseServer`: schema-v0 bridge for snapshot, put, and delete
- `Client` and `Server` schema-v0 dispatch hooks
- `CultNetHost.Database` and `CultNetHost.DatabaseServer`
- subscription request/cancel/change messages
- server-side live change fanout for schema/key-filtered subscriptions
Not implemented yet:
- remote subscription request/stream messages
- shard catalog exchange
- membership/failure detection
- leader election or automatic shard failover
@ -61,14 +62,13 @@ Not implemented yet:
## Next Coherent Slice
Add explicit subscription messages:
Add shard catalog exchange:
- subscribe by schema
- subscribe by record key
- subscribe by name/index
- unsubscribe
- stream domain changes as raw document put/delete records
- advertise shard ids, schema filters, key ranges/prefixes, owner runtime ids,
and epochs
- let clients ask any node where a write belongs
- reject stale-epoch writes explicitly
- prepare forwarding without making forwarding mandatory
That gives clients the RethinkDB/Firebase feel without requiring consensus or
membership work first. Shard catalogs and membership should follow once there
is more than one authority to discover.
That gives the subscription surface a real routing model before membership,
leader election, or replicated logs enter the room and start charging rent.

9
src/GameCult.Networking/CultNetDatabase.cs
View file

@ -390,6 +390,15 @@ namespace GameCult.Networking
.Select(change => (CultNetDatabaseChange<T>)change);
}
/// <summary>
/// Watches all typed domain changes as boxed change objects.
/// </summary>
public Observable<object> WatchAllChanges()
{
ThrowIfDisposed();
return _changes;
}
/// <summary>
/// Watches one record.
/// </summary>

150
src/GameCult.Networking/CultNetDatabaseServer.cs
View file

@ -1,6 +1,11 @@
using System;
using System.Collections.Concurrent;
using System.Linq;
using System.Reflection;
using System.Threading.Tasks;
using GameCult.Caching;
using LiteNetLib;
using R3;
namespace GameCult.Networking
{
@ -14,6 +19,9 @@ namespace GameCult.Networking
private readonly Func<CultNetSnapshotRequestMessage, NetPeer, Task> _snapshotHandler;
private readonly Func<CultNetDocumentPutRawMessage, NetPeer, Task> _putHandler;
private readonly Func<CultNetDocumentDeleteMessage, NetPeer, Task> _deleteHandler;
private readonly Func<CultNetDatabaseSubscribeMessage, NetPeer, Task> _subscribeHandler;
private readonly Func<CultNetDatabaseUnsubscribeMessage, NetPeer, Task> _unsubscribeHandler;
private readonly ConcurrentDictionary<string, IDisposable> _subscriptions = new(StringComparer.Ordinal);
private bool _disposed;
/// <summary>
@ -26,10 +34,14 @@ namespace GameCult.Networking
_snapshotHandler = HandleSnapshotRequestAsync;
_putHandler = HandlePutAsync;
_deleteHandler = HandleDeleteAsync;
_subscribeHandler = HandleSubscribeAsync;
_unsubscribeHandler = HandleUnsubscribeAsync;
_server.OnCultNet(_snapshotHandler);
_server.OnCultNet(_putHandler);
_server.OnCultNet(_deleteHandler);
_server.OnCultNet(_subscribeHandler);
_server.OnCultNet(_unsubscribeHandler);
}
/// <summary>
@ -79,6 +91,14 @@ namespace GameCult.Networking
_server.RemoveCultNetMessageListener<CultNetSnapshotRequestMessage>(_snapshotHandler);
_server.RemoveCultNetMessageListener<CultNetDocumentPutRawMessage>(_putHandler);
_server.RemoveCultNetMessageListener<CultNetDocumentDeleteMessage>(_deleteHandler);
_server.RemoveCultNetMessageListener<CultNetDatabaseSubscribeMessage>(_subscribeHandler);
_server.RemoveCultNetMessageListener<CultNetDatabaseUnsubscribeMessage>(_unsubscribeHandler);
foreach (var subscription in _subscriptions.Values)
{
subscription.Dispose();
}
_subscriptions.Clear();
}
private Task HandleSnapshotRequestAsync(CultNetSnapshotRequestMessage request, NetPeer peer)
@ -110,5 +130,135 @@ namespace GameCult.Networking
peer.SendCultNet(new CultNetErrorMessage { Error = ex.Message });
}
}
private Task HandleSubscribeAsync(CultNetDatabaseSubscribeMessage message, NetPeer peer)
{
var subscriptionId = string.IsNullOrWhiteSpace(message.SubscriptionId)
? message.MessageId
: message.SubscriptionId;
if (string.IsNullOrWhiteSpace(subscriptionId))
{
peer.SendCultNet(new CultNetErrorMessage { Error = "Database subscription requires a subscriptionId or messageId." });
return Task.CompletedTask;
}
var key = SubscriptionKey(peer, subscriptionId);
_subscriptions.AddOrUpdate(
key,
_ => CreateSubscription(message, subscriptionId, peer),
(_, existing) =>
{
existing.Dispose();
return CreateSubscription(message, subscriptionId, peer);
});
if (message.IncludeSnapshot)
{
peer.SendCultNet(CreateSnapshotResponse(new CultNetSnapshotRequestMessage
{
MessageId = message.MessageId,
SchemaIds = message.SchemaIds,
RecordKeys = message.RecordKeys
}));
}
return Task.CompletedTask;
}
private Task HandleUnsubscribeAsync(CultNetDatabaseUnsubscribeMessage message, NetPeer peer)
{
var subscriptionId = string.IsNullOrWhiteSpace(message.SubscriptionId)
? message.MessageId
: message.SubscriptionId;
if (!string.IsNullOrWhiteSpace(subscriptionId) &&
_subscriptions.TryRemove(SubscriptionKey(peer, subscriptionId), out var subscription))
{
subscription.Dispose();
}
return Task.CompletedTask;
}
private IDisposable CreateSubscription(
CultNetDatabaseSubscribeMessage request,
string subscriptionId,
NetPeer peer)
{
return _database.WatchAllChanges().Subscribe(change =>
{
var outbound = CreateChangeMessage(change, subscriptionId, request);
if (outbound != null)
{
peer.SendCultNet(outbound);
}
});
}
internal CultNetDatabaseChangeRawMessage? CreateChangeMessage(
object change,
string subscriptionId,
CultNetDatabaseSubscribeMessage request)
{
var changeType = change.GetType();
var key = (CultRecordKey)(changeType.GetProperty("Key")?.GetValue(change) ?? new CultRecordKey(string.Empty));
var schemaId = (string?)changeType.GetProperty("SchemaId")?.GetValue(change) ?? string.Empty;
if (!Matches(request, schemaId, key))
{
return null;
}
var kind = (CultNetDatabaseChangeKind)(changeType.GetProperty("Kind")?.GetValue(change) ?? CultNetDatabaseChangeKind.Updated);
var document = changeType.GetProperty("Document")?.GetValue(change);
if (kind == CultNetDatabaseChangeKind.Removed || document == null)
{
return new CultNetDatabaseChangeRawMessage
{
MessageId = Guid.NewGuid().ToString("N"),
SubscriptionId = subscriptionId,
ChangeKind = "removed",
RecordKey = key.Value,
SchemaId = schemaId
};
}
return new CultNetDatabaseChangeRawMessage
{
MessageId = Guid.NewGuid().ToString("N"),
SubscriptionId = subscriptionId,
ChangeKind = kind == CultNetDatabaseChangeKind.Added ? "added" : "updated",
Document = CreateRawDocumentRecord(key, document)
};
}
private CultNetRawDocumentRecord CreateRawDocumentRecord(CultRecordKey key, object document)
{
var method = typeof(CultNetDocumentRegistry)
.GetMethods(BindingFlags.Public | BindingFlags.Instance)
.Single(candidate => candidate.Name == nameof(CultNetDocumentRegistry.CreateRawDocumentPutMessage) &&
candidate.IsGenericMethodDefinition);
var documentType = document.GetType();
var handleType = typeof(CultRecordHandle<>).MakeGenericType(documentType);
var handle = Activator.CreateInstance(handleType, new object[] { key });
var message = method
.MakeGenericMethod(documentType)
.Invoke(_database.Documents, new[] { Guid.NewGuid().ToString("N"), handle, document, null });
return ((CultNetDocumentPutRawMessage)message!).Document;
}
private static bool Matches(CultNetDatabaseSubscribeMessage request, string schemaId, CultRecordKey key)
{
var schemaMatches = request.SchemaIds == null ||
request.SchemaIds.Length == 0 ||
request.SchemaIds.Contains(schemaId, StringComparer.Ordinal);
var keyMatches = request.RecordKeys == null ||
request.RecordKeys.Length == 0 ||
request.RecordKeys.Contains(key.Value, StringComparer.Ordinal);
return schemaMatches && keyMatches;
}
private static string SubscriptionKey(NetPeer peer, string subscriptionId)
{
return $"{peer.Id}:{subscriptionId}";
}
}
}

3
src/GameCult.Networking/CultNetSchemaMessageSerialization.cs
View file

@ -59,6 +59,9 @@ namespace GameCult.Networking
CultNetSchemaVersions.DocumentPutRaw => Deserialize<CultNetDocumentPutRawMessage>(payload),
CultNetSchemaVersions.SnapshotRequest => Deserialize<CultNetSnapshotRequestMessage>(payload),
CultNetSchemaVersions.SnapshotResponseRaw => Deserialize<CultNetSnapshotResponseRawMessage>(payload),
CultNetSchemaVersions.DatabaseSubscribe => Deserialize<CultNetDatabaseSubscribeMessage>(payload),
CultNetSchemaVersions.DatabaseUnsubscribe => Deserialize<CultNetDatabaseUnsubscribeMessage>(payload),
CultNetSchemaVersions.DatabaseChangeRaw => Deserialize<CultNetDatabaseChangeRawMessage>(payload),
CultNetSchemaVersions.SchemaCatalogRequest => Deserialize<CultNetSchemaCatalogRequestMessage>(payload),
CultNetSchemaVersions.SchemaCatalogResponse => Deserialize<CultNetSchemaCatalogResponseMessage>(payload),
_ => throw new MessagePackSerializationException(

100
src/GameCult.Networking/CultNetSchemaMessages.cs
View file

@ -79,6 +79,18 @@ namespace GameCult.Networking
/// schema catalog response contract identifier.
/// </summary>
public const string SchemaCatalogResponse = "cultnet.schema_catalog_response.v0";
/// <summary>
/// database subscription request contract identifier.
/// </summary>
public const string DatabaseSubscribe = "cultnet.database_subscribe.v0";
/// <summary>
/// database subscription cancel contract identifier.
/// </summary>
public const string DatabaseUnsubscribe = "cultnet.database_unsubscribe.v0";
/// <summary>
/// database subscription change contract identifier.
/// </summary>
public const string DatabaseChangeRaw = "cultnet.database_change_raw.v0";
}
/// <summary>
@ -475,6 +487,94 @@ namespace GameCult.Networking
[Key("documents")] public CultNetRawDocumentRecord[] Documents { get; set; } = Array.Empty<CultNetRawDocumentRecord>();
}
/// <summary>
/// Requests a live database subscription.
/// </summary>
[MessagePackObject]
public class CultNetDatabaseSubscribeMessage : ICultNetSchemaMessage
{
/// <summary>
/// Gets or sets the schema version.
/// </summary>
[Key("schemaVersion")] public string SchemaVersion { get; set; } = CultNetSchemaVersions.DatabaseSubscribe;
/// <summary>
/// Gets or sets the request id.
/// </summary>
[Key("messageId")] public string MessageId { get; set; } = string.Empty;
/// <summary>
/// Gets or sets the live subscription id.
/// </summary>
[Key("subscriptionId")] public string SubscriptionId { get; set; } = string.Empty;
/// <summary>
/// Gets or sets optional schema ids to watch.
/// </summary>
[Key("schemaIds")] public string[]? SchemaIds { get; set; }
/// <summary>
/// Gets or sets optional record keys to watch.
/// </summary>
[Key("recordKeys")] public string[]? RecordKeys { get; set; }
/// <summary>
/// Gets or sets whether the server should send a matching snapshot before live changes.
/// </summary>
[Key("includeSnapshot")] public bool IncludeSnapshot { get; set; } = true;
}
/// <summary>
/// Cancels a live database subscription.
/// </summary>
[MessagePackObject]
public class CultNetDatabaseUnsubscribeMessage : ICultNetSchemaMessage
{
/// <summary>
/// Gets or sets the schema version.
/// </summary>
[Key("schemaVersion")] public string SchemaVersion { get; set; } = CultNetSchemaVersions.DatabaseUnsubscribe;
/// <summary>
/// Gets or sets the request id.
/// </summary>
[Key("messageId")] public string MessageId { get; set; } = string.Empty;
/// <summary>
/// Gets or sets the live subscription id.
/// </summary>
[Key("subscriptionId")] public string SubscriptionId { get; set; } = string.Empty;
}
/// <summary>
/// Streams one raw document change for a live database subscription.
/// </summary>
[MessagePackObject]
public class CultNetDatabaseChangeRawMessage : ICultNetSchemaMessage
{
/// <summary>
/// Gets or sets the schema version.
/// </summary>
[Key("schemaVersion")] public string SchemaVersion { get; set; } = CultNetSchemaVersions.DatabaseChangeRaw;
/// <summary>
/// Gets or sets the message id.
/// </summary>
[Key("messageId")] public string MessageId { get; set; } = string.Empty;
/// <summary>
/// Gets or sets the live subscription id.
/// </summary>
[Key("subscriptionId")] public string SubscriptionId { get; set; } = string.Empty;
/// <summary>
/// Gets or sets the change kind.
/// </summary>
[Key("changeKind")] public string ChangeKind { get; set; } = string.Empty;
/// <summary>
/// Gets or sets the current document for added or updated changes.
/// </summary>
[Key("document")] public CultNetRawDocumentRecord? Document { get; set; }
/// <summary>
/// Gets or sets the deleted record key for removed changes.
/// </summary>
[Key("recordKey")] public string? RecordKey { get; set; }
/// <summary>
/// Gets or sets the deleted schema id for removed changes.
/// </summary>
[Key("schemaId")] public string? SchemaId { get; set; }
}
/// <summary>
/// Describes one schema exposed through the CultNet schema catalog.
/// </summary>

5
src/GameCult.Networking/README.md
View file

@ -131,8 +131,9 @@ CultNet becomes the distribution model:
The current first slice exposes this as `CultNetDatabase` and wires hosted
servers through `CultNetDatabaseServer`. It handles primary-shard writes, typed
R3 watch streams, raw document puts/deletes, and raw snapshot responses. Remote
subscription messages and shard catalog exchange are the next missing organs.
R3 watch streams, raw document puts/deletes, raw snapshot responses, and
schema/key-filtered live subscriptions. Shard catalog exchange is the next
missing organ.
The first coherent shard policy is primary ownership. Each shard has one
authoritative writer at a time. Clients may connect to any node; a non-owner

81
tests/GameCult.Networking.Tests/NetworkingTests.cs
View file

@ -214,6 +214,33 @@ namespace GameCult.Networking.Tests
Assert.That(roundTrip.Documents[0].Tags, Is.EqualTo(["swarm", "dream"]));
}
[Test]
public void CultNetSchemaMessageSerialization_RoundTrips_DatabaseChangeRaw()
{
var message = new CultNetDatabaseChangeRawMessage
{
MessageId = "change-1",
SubscriptionId = "sub-1",
ChangeKind = "updated",
Document = new CultNetRawDocumentRecord
{
SchemaId = "schema-1",
RecordKey = "record-1",
StoredAt = "2026-05-19T12:00:00.0000000+00:00",
PayloadEncoding = "messagepack",
Payload = [0x91, 0x01]
}
};
var payload = CultNetSchemaMessageSerialization.Serialize(message);
var roundTrip = (CultNetDatabaseChangeRawMessage)CultNetSchemaMessageSerialization.Deserialize(payload);
Assert.That(roundTrip.SubscriptionId, Is.EqualTo("sub-1"));
Assert.That(roundTrip.ChangeKind, Is.EqualTo("updated"));
Assert.That(roundTrip.Document, Is.Not.Null);
Assert.That(roundTrip.Document!.RecordKey, Is.EqualTo("record-1"));
}
[Test]
public async Task CultNetDocumentRegistry_RawSnapshotReplication_PreservesPayloadBytes()
{
@ -389,6 +416,60 @@ namespace GameCult.Networking.Tests
Assert.That(response.Documents[0].Payload, Is.EqualTo(SerializePlayerDataPayload(player)));
}
[Test]
public void CultNetDatabaseServer_Creates_Filtered_SubscriptionChange()
{
var cache = new CultCache();
var registry = new CultNetDocumentRegistry(cache.Registry)
.Register(CultNetDocumentBinding.ForDocument<PlayerData>(
cache.Registry,
payloadSerializer: SerializePlayerDataPayload,
payloadDeserializer: DeserializePlayerDataPayload));
var database = new CultNetDatabase(cache, new CultNetDatabaseOptions
{
DocumentRegistry = registry
});
using var server = new Server(cache, DevelopmentServerSecurity);
using var databaseServer = new CultNetDatabaseServer(server, database);
var key = new CultRecordKey("player:change");
var player = new PlayerData
{
PlayerId = Guid.NewGuid(),
Email = "change@example.test",
PasswordHash = "hash",
Username = "Change"
};
var schemaId = cache.Registry.GetRequired<PlayerData>().SchemaId;
var change = new CultNetDatabaseChange<PlayerData>(
CultNetDatabaseChangeKind.Added,
key,
schemaId,
database.Shards[0],
player,
previousDocument: null);
var method = typeof(CultNetDatabaseServer).GetMethod(
"CreateChangeMessage",
BindingFlags.Instance | BindingFlags.NonPublic)!;
var message = (CultNetDatabaseChangeRawMessage?)method.Invoke(databaseServer, new object[]
{
change,
"sub-1",
new CultNetDatabaseSubscribeMessage
{
SubscriptionId = "sub-1",
SchemaIds = [schemaId],
RecordKeys = [key.Value]
}
});
Assert.That(message, Is.Not.Null);
Assert.That(message!.SubscriptionId, Is.EqualTo("sub-1"));
Assert.That(message.ChangeKind, Is.EqualTo("added"));
Assert.That(message.Document, Is.Not.Null);
Assert.That(message.Document!.Payload, Is.EqualTo(SerializePlayerDataPayload(player)));
}
[Test]
public void CultNetSchemaRegistry_BuiltInCatalog_AdvertisesRawLane_AndSharedGhostlightContract()
{