From 9fa85d755fb5856acc3412f3b1fab7167b249d18 Mon Sep 17 00:00:00 2001 From: Meta Vens Date: Tue, 19 May 2026 18:48:53 +0100 Subject: [PATCH] Add CultNet simulation observation server --- .../cultnet-distributed-database/summary.md | 3 +- .../Contracts/cultnet-distributed-database.md | 6 +- .../CultNetSimulationObservationServer.cs | 79 +++++++++++++++++++ src/GameCult.Networking/README.md | 3 + .../NetworkingTests.cs | 50 ++++++++++++ 5 files changed, 138 insertions(+), 3 deletions(-) create mode 100644 src/GameCult.Networking/CultNetSimulationObservationServer.cs diff --git a/research/cultnet-distributed-database/summary.md b/research/cultnet-distributed-database/summary.md index f85cee62..4d247b0d 100644 --- a/research/cultnet-distributed-database/summary.md +++ b/research/cultnet-distributed-database/summary.md @@ -176,6 +176,7 @@ Shard catalog exchange now exists: - simulation witness observations and deterministic consensus candidates - schema-v0 simulation observation and candidate messages - reactive observation hub for witness gossip and consensus candidates +- server-side observation bridge for observation messages and candidate replies ## Next Cut @@ -186,7 +187,7 @@ Build durable authoritative mutation logs: - return explicit resync-required responses when requested history is gone - then add simulation-frame rollback and resimulation helpers on top of predicted input streams -- add network fanout for observation gossip and candidate propagation +- add peer-to-peer fanout for observation gossip and candidate propagation That is the next foundation needed before membership, Raft-style failover, or CRDT policy work can be honest. diff --git a/src/GameCult.Networking/Contracts/cultnet-distributed-database.md b/src/GameCult.Networking/Contracts/cultnet-distributed-database.md index 4b5dc69c..ae5e1410 100644 --- a/src/GameCult.Networking/Contracts/cultnet-distributed-database.md +++ b/src/GameCult.Networking/Contracts/cultnet-distributed-database.md @@ -66,6 +66,8 @@ Implemented: - schema-v0 simulation observation and consensus candidate messages - reactive simulation observation hub that receives witness gossip and emits consensus candidates +- server-side observation bridge that accepts observation messages and replies + with current candidate messages Not implemented yet: @@ -73,7 +75,7 @@ Not implemented yet: - leader election or automatic shard failover - durable log persistence and compaction - snapshot fallback for compacted log history -- network fanout for observation gossip and candidate propagation +- peer-to-peer fanout for observation gossip and candidate propagation - rollback/resimulation helpers for simulation frames - declared CRDT merge policies @@ -108,7 +110,7 @@ Add durable mutation-log storage and snapshot fallback: - return `ResyncRequired` when a replica asks for compacted history - after that, add simulation-frame rollback/resimulation helpers around predicted input streams -- add network fanout for observation gossip and candidate propagation +- add peer-to-peer fanout for observation gossip and candidate propagation The log is wire-readable, replicas can apply it explicitly, a pull loop can drive non-primary shards from primary endpoints, and replica cursors can survive diff --git a/src/GameCult.Networking/CultNetSimulationObservationServer.cs b/src/GameCult.Networking/CultNetSimulationObservationServer.cs new file mode 100644 index 00000000..2b301197 --- /dev/null +++ b/src/GameCult.Networking/CultNetSimulationObservationServer.cs @@ -0,0 +1,79 @@ +using System; +using System.Linq; +using System.Threading.Tasks; +using LiteNetLib; + +namespace GameCult.Networking +{ + /// + /// Bridges schema-v0 simulation observation messages into a local observation hub. + /// + public sealed class CultNetSimulationObservationServer : IDisposable + { + private readonly Server _server; + private readonly CultNetSimulationObservationHub _hub; + private readonly Func _observationHandler; + private bool _disposed; + + /// + /// Creates and attaches a simulation observation bridge. + /// + public CultNetSimulationObservationServer( + Server server, + CultNetSimulationObservationHub hub) + { + _server = server ?? throw new ArgumentNullException(nameof(server)); + _hub = hub ?? throw new ArgumentNullException(nameof(hub)); + _observationHandler = HandleObservationAsync; + _server.OnCultNet(_observationHandler); + } + + /// + /// Gets the observation hub used by this bridge. + /// + public CultNetSimulationObservationHub Hub => _hub; + + /// + /// Applies an observation message and returns current candidate messages for the same subject. + /// + public CultNetSimulationConsensusCandidateMessage[] CreateCandidateMessages( + CultNetSimulationObservationMessage message) + { + if (message == null) throw new ArgumentNullException(nameof(message)); + return _hub.Submit(message) + .Select(candidate => CultNetSimulationConsensusCandidateMessage.FromCandidate( + message.MessageId, + candidate)) + .ToArray(); + } + + /// + public void Dispose() + { + if (_disposed) + { + return; + } + + _disposed = true; + _server.RemoveCultNetMessageListener(_observationHandler); + } + + private Task HandleObservationAsync(CultNetSimulationObservationMessage message, NetPeer peer) + { + try + { + foreach (var candidate in CreateCandidateMessages(message)) + { + peer.SendCultNet(candidate); + } + } + catch (Exception ex) + { + peer.SendCultNet(new CultNetErrorMessage { Error = ex.Message }); + } + + return Task.CompletedTask; + } + } +} diff --git a/src/GameCult.Networking/README.md b/src/GameCult.Networking/README.md index a4b4b6b1..f518cae1 100644 --- a/src/GameCult.Networking/README.md +++ b/src/GameCult.Networking/README.md @@ -163,6 +163,9 @@ path, not committed world state by themselves. The schema-v0 wire contracts are `cultnet.simulation_observation.v0` and `cultnet.simulation_consensus_candidate.v0`. `CultNetSimulationObservationHub` collects incoming observations and emits candidate updates reactively. +`CultNetSimulationObservationServer` attaches that hub to a server so incoming +observation messages update the hub and receive current candidate messages in +reply. 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 diff --git a/tests/GameCult.Networking.Tests/NetworkingTests.cs b/tests/GameCult.Networking.Tests/NetworkingTests.cs index 19b044f6..b63d9971 100644 --- a/tests/GameCult.Networking.Tests/NetworkingTests.cs +++ b/tests/GameCult.Networking.Tests/NetworkingTests.cs @@ -1431,6 +1431,56 @@ namespace GameCult.Networking.Tests Assert.That(candidates.Exists(candidate => candidate.HasQuorum), Is.True); } + [Test] + public void CultNetSimulationObservationServer_Creates_CandidateMessages() + { + var cache = new CultCache(); + using var server = new Server(cache, DevelopmentServerSecurity); + using var hub = new CultNetSimulationObservationHub(new CultNetSimulationConsensusOptions + { + MinimumWitnesses = 2, + QuorumRatio = 1d + }); + using var observationServer = new CultNetSimulationObservationServer(server, hub); + var claimHash = CultNetSimulationObservation.ComputeClaimHash("hit", "alice", "bob", "frame:100"); + var first = new CultNetSimulationObservationMessage + { + MessageId = "observe-1", + Observation = new CultNetSimulationObservation + { + WitnessRuntimeId = "watcher-1", + ShardId = "arena", + ShardEpoch = 1, + Frame = 100, + SubjectId = "bob", + ClaimKind = "hit", + ClaimHash = claimHash + } + }; + var second = new CultNetSimulationObservationMessage + { + MessageId = "observe-2", + Observation = new CultNetSimulationObservation + { + WitnessRuntimeId = "watcher-2", + ShardId = "arena", + ShardEpoch = 1, + Frame = 100, + SubjectId = "bob", + ClaimKind = "hit", + ClaimHash = claimHash + } + }; + + observationServer.CreateCandidateMessages(first); + var candidates = observationServer.CreateCandidateMessages(second); + + Assert.That(candidates, Has.Length.EqualTo(1)); + Assert.That(candidates[0].MessageId, Is.EqualTo("observe-2")); + Assert.That(candidates[0].ClaimHash, Is.EqualTo(claimHash)); + Assert.That(candidates[0].HasQuorum, Is.True); + } + [Test] public void CultNetDatabaseServer_Creates_Filtered_SubscriptionChange() {