Isolate realtime provider broadcasts from departed peers
This commit is contained in:
parent
1d126f4a35
commit
8cac8c9ec7
17 changed files with 68 additions and 4 deletions
|
|
@ -217,10 +217,30 @@ public sealed class CultMeshQuicRealtimeServer : IAsyncDisposable
|
|||
CancellationToken cancellationToken = default)
|
||||
{
|
||||
if (frame == null) throw new ArgumentNullException(nameof(frame));
|
||||
var sends = _clients.Keys.Select(client => client.SendAsync(frame, cancellationToken)).ToArray();
|
||||
var sends = _clients.Keys.Select(client => SendToConnectedClientAsync(client, frame, cancellationToken)).ToArray();
|
||||
if (sends.Length > 0) await Task.WhenAll(sends).ConfigureAwait(false);
|
||||
}
|
||||
|
||||
private async Task SendToConnectedClientAsync(
|
||||
CultMeshQuicRealtimeTransport client,
|
||||
CultMeshRealtimeFrame frame,
|
||||
CancellationToken cancellationToken)
|
||||
{
|
||||
try
|
||||
{
|
||||
await client.SendAsync(frame, cancellationToken).ConfigureAwait(false);
|
||||
}
|
||||
catch (OperationCanceledException) when (cancellationToken.IsCancellationRequested)
|
||||
{
|
||||
throw;
|
||||
}
|
||||
catch
|
||||
{
|
||||
if (_clients.TryRemove(client, out _))
|
||||
client.Dispose();
|
||||
}
|
||||
}
|
||||
|
||||
/// <summary>Receives the next client-originated state frame.</summary>
|
||||
public Task<CultMeshRealtimeFrame> ReceiveAsync(CancellationToken cancellationToken = default) =>
|
||||
_received.ReceiveAsync(cancellationToken);
|
||||
|
|
|
|||
|
|
@ -49,6 +49,10 @@ advertised mapping when it is actually reachable; otherwise it connects to the
|
|||
advertised QUIC route. Latest-only receivers must replace pending generations
|
||||
per body rather than queueing render debt.
|
||||
|
||||
Provider broadcast isolates physical peers. A client disconnect or send failure
|
||||
removes that session without failing the publication tick for healthy clients;
|
||||
only caller cancellation stops the broadcast operation.
|
||||
|
||||
**Derived state:** connector choice, physical endpoint, connection health, and
|
||||
transport diagnostics. None of these grant provider authority or become content
|
||||
identity.
|
||||
|
|
|
|||
|
|
@ -78,6 +78,46 @@ public sealed class CultMeshNativeQuicRealtimeTransportTests
|
|||
.WithMessage("*certificate*pin*");
|
||||
}
|
||||
|
||||
[Test]
|
||||
public async Task NativeClientDepartureCannotBreakProviderBroadcast()
|
||||
{
|
||||
if (!OperatingSystem.IsWindows() || !QuicListener.IsSupported)
|
||||
Assert.Ignore("Native MsQuic integration requires Windows QUIC support.");
|
||||
using var certificate = CreateCertificate();
|
||||
await using var server = await CultMeshQuicRealtimeServer.ListenAsync(new CultMeshQuicRealtimeServerOptions
|
||||
{
|
||||
ListenEndPoint = new IPEndPoint(IPAddress.Loopback, 0),
|
||||
ServerCertificate = certificate
|
||||
});
|
||||
var pin = Convert.ToHexString(SHA256.HashData(certificate.RawData));
|
||||
var endpoint = $"cultmesh-state+quic://127.0.0.1:{server.LocalEndPoint.Port}?cert-sha256={pin}";
|
||||
var connector = new CultMeshNativeQuicRealtimeTransportConnector();
|
||||
var client = await connector.ConnectAsync(
|
||||
new CultMeshTransportCandidate(endpoint),
|
||||
CultMeshEndpointId.Parse("aetheria.local"));
|
||||
await WaitUntilAsync(() => server.ConnectionCount == 1);
|
||||
await server.BroadcastAsync(Frame(sequence: 1));
|
||||
using var receiveTimeout = new CancellationTokenSource(TimeSpan.FromSeconds(5));
|
||||
await client.ReceiveAsync(receiveTimeout.Token);
|
||||
|
||||
client.Dispose();
|
||||
|
||||
Func<Task> broadcastAfterDeparture = () => server.BroadcastAsync(Frame(sequence: 2));
|
||||
await broadcastAfterDeparture.Should().NotThrowAsync(
|
||||
"a short-lived realtime observer cannot terminate provider publication");
|
||||
}
|
||||
|
||||
private static CultMeshRealtimeFrame Frame(long sequence) => new()
|
||||
{
|
||||
ChannelId = "aetheria.entities",
|
||||
SchemaId = "gamecult.aetheria.entity_soa.v1",
|
||||
BodyId = "aetheria.entities.current-zone",
|
||||
ProducerEpoch = 17,
|
||||
Sequence = sequence,
|
||||
Delivery = CultMeshRealtimeDelivery.LatestOnly,
|
||||
Payload = new byte[] { 1, 3, 3, 7 }
|
||||
};
|
||||
|
||||
private static async Task WaitUntilAsync(Func<bool> predicate)
|
||||
{
|
||||
using var timeout = new CancellationTokenSource(TimeSpan.FromSeconds(5));
|
||||
|
|
|
|||
|
|
@ -9,10 +9,10 @@ runtime; the Microsoft license is included under `Third Party Notices`.
|
|||
Build it with `scripts/build-unity-package.ps1`. Do not edit generated plugin
|
||||
assemblies in the package output.
|
||||
|
||||
Install version `1.0.36` through Unity Package Manager with:
|
||||
Install version `1.0.37` through Unity Package Manager with:
|
||||
|
||||
```text
|
||||
https://github.com/GameCult/CultLib.git?path=/unity/org.gamecult.cultlib#cultlib-unity-v1.0.36
|
||||
https://github.com/GameCult/CultLib.git?path=/unity/org.gamecult.cultlib#cultlib-unity-v1.0.37
|
||||
```
|
||||
|
||||
CultMesh chooses mapped memory for reachable same-machine bodies and advertised
|
||||
|
|
|
|||
Binary file not shown.
Binary file not shown.
Binary file not shown.
Binary file not shown.
Binary file not shown.
Binary file not shown.
Binary file not shown.
Binary file not shown.
Binary file not shown.
Binary file not shown.
Binary file not shown.
Binary file not shown.
|
|
@ -1,7 +1,7 @@
|
|||
{
|
||||
"name": "org.gamecult.cultlib",
|
||||
"displayName": "CultLib",
|
||||
"version": "1.0.36",
|
||||
"version": "1.0.37",
|
||||
"unity": "2021.3",
|
||||
"description": "CultCache, CultNet, and CultMesh managed runtime assemblies for Unity.",
|
||||
"license": "MIT",
|
||||
|
|
|
|||
Loading…
Reference in a new issue