diff --git a/research/cultnet-distributed-database/summary.md b/research/cultnet-distributed-database/summary.md
index 40a0b19d..f85cee62 100644
--- a/research/cultnet-distributed-database/summary.md
+++ b/research/cultnet-distributed-database/summary.md
@@ -175,6 +175,7 @@ Shard catalog exchange now exists:
- predicted/reconciled change events for client-side prediction
- simulation witness observations and deterministic consensus candidates
- schema-v0 simulation observation and candidate messages
+- reactive observation hub for witness gossip and consensus candidates
## Next Cut
@@ -185,7 +186,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 an observation hub that receives gossip and emits consensus candidates
+- add network 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 6cbaf72f..4b5dc69c 100644
--- a/src/GameCult.Networking/Contracts/cultnet-distributed-database.md
+++ b/src/GameCult.Networking/Contracts/cultnet-distributed-database.md
@@ -64,6 +64,8 @@ Implemented:
- simulation witness observations for mesh-side opinions about frame facts
- deterministic consensus candidates derived from witness observations
- schema-v0 simulation observation and consensus candidate messages
+- reactive simulation observation hub that receives witness gossip and emits
+ consensus candidates
Not implemented yet:
@@ -71,7 +73,7 @@ Not implemented yet:
- leader election or automatic shard failover
- durable log persistence and compaction
- snapshot fallback for compacted log history
-- observation hub/fanout that receives gossip and emits candidates reactively
+- network fanout for observation gossip and candidate propagation
- rollback/resimulation helpers for simulation frames
- declared CRDT merge policies
@@ -106,8 +108,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 an observation hub that receives schema-v0 observation messages and emits
- consensus candidates reactively
+- add network 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/CultNetSimulationConsensus.cs b/src/GameCult.Networking/CultNetSimulationConsensus.cs
index 96407581..a666c368 100644
--- a/src/GameCult.Networking/CultNetSimulationConsensus.cs
+++ b/src/GameCult.Networking/CultNetSimulationConsensus.cs
@@ -4,6 +4,7 @@ using System.Linq;
using System.Security.Cryptography;
using System.Text;
using MessagePack;
+using R3;
namespace GameCult.Networking
{
@@ -290,4 +291,109 @@ namespace GameCult.Networking
observation.ClaimKind);
}
}
+
+ ///
+ /// Collects witness observations and publishes updated consensus candidates.
+ ///
+ public sealed class CultNetSimulationObservationHub : IDisposable
+ {
+ private readonly List _observations = new();
+ private readonly CultNetSimulationConsensus _consensus;
+ private readonly Subject _observationStream = new();
+ private readonly Subject _candidateStream = new();
+ private bool _disposed;
+
+ ///
+ /// Creates an observation hub.
+ ///
+ public CultNetSimulationObservationHub(CultNetSimulationConsensusOptions? options = null)
+ {
+ _consensus = new CultNetSimulationConsensus(options);
+ }
+
+ ///
+ /// Gets all observations currently held by the hub.
+ ///
+ public IReadOnlyList Observations => _observations;
+
+ ///
+ /// Watches accepted witness observations.
+ ///
+ public Observable WatchObservations()
+ {
+ ThrowIfDisposed();
+ return _observationStream;
+ }
+
+ ///
+ /// Watches consensus candidate updates.
+ ///
+ public Observable WatchCandidates()
+ {
+ ThrowIfDisposed();
+ return _candidateStream;
+ }
+
+ ///
+ /// Adds an observation and returns current candidates for the same observed subject.
+ ///
+ public IReadOnlyList Submit(CultNetSimulationObservation observation)
+ {
+ ThrowIfDisposed();
+ if (observation == null) throw new ArgumentNullException(nameof(observation));
+ _observations.Add(observation);
+ _observationStream.OnNext(observation);
+
+ var candidates = _consensus.BuildCandidates(_observations)
+ .Where(candidate => MatchesSubject(candidate, observation))
+ .ToArray();
+ foreach (var candidate in candidates)
+ {
+ _candidateStream.OnNext(candidate);
+ }
+
+ return candidates;
+ }
+
+ ///
+ /// Adds an observation message and returns current candidates for the same observed subject.
+ ///
+ public IReadOnlyList Submit(CultNetSimulationObservationMessage message)
+ {
+ if (message == null) throw new ArgumentNullException(nameof(message));
+ return Submit(message.Observation);
+ }
+
+ ///
+ public void Dispose()
+ {
+ if (_disposed)
+ {
+ return;
+ }
+
+ _disposed = true;
+ _observationStream.Dispose();
+ _candidateStream.Dispose();
+ }
+
+ private static bool MatchesSubject(
+ CultNetSimulationConsensusCandidate candidate,
+ CultNetSimulationObservation observation)
+ {
+ return string.Equals(candidate.ShardId, observation.ShardId, StringComparison.Ordinal) &&
+ candidate.ShardEpoch == observation.ShardEpoch &&
+ candidate.Frame == observation.Frame &&
+ string.Equals(candidate.SubjectId, observation.SubjectId, StringComparison.Ordinal) &&
+ string.Equals(candidate.ClaimKind, observation.ClaimKind, StringComparison.Ordinal);
+ }
+
+ private void ThrowIfDisposed()
+ {
+ if (_disposed)
+ {
+ throw new ObjectDisposedException(nameof(CultNetSimulationObservationHub));
+ }
+ }
+ }
}
diff --git a/src/GameCult.Networking/README.md b/src/GameCult.Networking/README.md
index d1edf3f1..a4b4b6b1 100644
--- a/src/GameCult.Networking/README.md
+++ b/src/GameCult.Networking/README.md
@@ -161,7 +161,8 @@ into candidates with support weight, total observed weight, confidence, and
quorum status. Those candidates are opinions ready for an authoritative commit
path, not committed world state by themselves. The schema-v0 wire contracts are
`cultnet.simulation_observation.v0` and
-`cultnet.simulation_consensus_candidate.v0`.
+`cultnet.simulation_consensus_candidate.v0`. `CultNetSimulationObservationHub`
+collects incoming observations and emits candidate updates reactively.
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 1d87efae..19b044f6 100644
--- a/tests/GameCult.Networking.Tests/NetworkingTests.cs
+++ b/tests/GameCult.Networking.Tests/NetworkingTests.cs
@@ -1385,6 +1385,52 @@ namespace GameCult.Networking.Tests
Assert.That(candidates[0].HasQuorum, Is.True);
}
+ [Test]
+ public void CultNetSimulationObservationHub_Emits_ReactiveCandidates()
+ {
+ var claimHash = CultNetSimulationObservation.ComputeClaimHash("hit", "alice", "bob", "frame:100");
+ using var hub = new CultNetSimulationObservationHub(new CultNetSimulationConsensusOptions
+ {
+ MinimumWitnesses = 2,
+ QuorumRatio = 1d
+ });
+ var observations = new List();
+ var candidates = new List();
+ using var observationSubscription = hub.WatchObservations()
+ .Subscribe(observation => observations.Add(observation));
+ using var candidateSubscription = hub.WatchCandidates()
+ .Subscribe(candidate => candidates.Add(candidate));
+
+ hub.Submit(new CultNetSimulationObservation
+ {
+ WitnessRuntimeId = "watcher-1",
+ ShardId = "arena",
+ ShardEpoch = 1,
+ Frame = 100,
+ SubjectId = "bob",
+ ClaimKind = "hit",
+ ClaimHash = claimHash
+ });
+ var current = hub.Submit(new CultNetSimulationObservationMessage
+ {
+ Observation = new CultNetSimulationObservation
+ {
+ WitnessRuntimeId = "watcher-2",
+ ShardId = "arena",
+ ShardEpoch = 1,
+ Frame = 100,
+ SubjectId = "bob",
+ ClaimKind = "hit",
+ ClaimHash = claimHash
+ }
+ });
+
+ Assert.That(observations, Has.Count.EqualTo(2));
+ Assert.That(current, Has.Count.EqualTo(1));
+ Assert.That(current[0].HasQuorum, Is.True);
+ Assert.That(candidates.Exists(candidate => candidate.HasQuorum), Is.True);
+ }
+
[Test]
public void CultNetDatabaseServer_Creates_Filtered_SubscriptionChange()
{