diff --git a/src/Mimir.BufferSmoke/Program.cs b/src/Mimir.BufferSmoke/Program.cs index ac207a7..b21165a 100644 --- a/src/Mimir.BufferSmoke/Program.cs +++ b/src/Mimir.BufferSmoke/Program.cs @@ -134,6 +134,12 @@ if (args.Any(arg => string.Equals(arg, "--muninn-move-cultmesh-stream-smoke", St ParseStringOption(args, "--native-reservoir", DefaultNativeReservoirPath())); } +if (args.Any(arg => string.Equals(arg, "--muninn-move-odin-live-smoke", StringComparison.OrdinalIgnoreCase))) +{ + return RunMuninnMoveOdinLiveSmoke( + ParseStringOption(args, "--odin-cultmesh-uri", "cultmesh://127.0.0.1:17871/rendezvous/provider-catalog")); +} + if (args.Any(arg => string.Equals(arg, "--move-fusion-smoke", StringComparison.OrdinalIgnoreCase))) { return RunMoveFusionSmoke(); @@ -192,7 +198,16 @@ if (args.Any(arg => string.Equals(arg, "--move-proof-runtime-snapshot-smoke", St ParseStringOption(args, "--native-reservoir", DefaultNativeReservoirPath()), activateViaConfiguredSource: true, requireActivationSurface: true, - activateViaSnapshotProvider: true); + activateViaSnapshotProvider: true, + snapshotOutputPath: ParseStringOption(args, "--snapshot-output", "")); +} + +if (args.Any(arg => string.Equals(arg, "--move-proof-runtime-snapshot-file-smoke", StringComparison.OrdinalIgnoreCase))) +{ + return RunMoveProofRuntimeSnapshotFileSmoke( + ParseStringOption(args, "--native-reservoir", DefaultNativeReservoirPath()), + ParseStringOption(args, "--snapshot", ""), + args.Any(arg => string.Equals(arg, "--require-full-pose", StringComparison.OrdinalIgnoreCase))); } if (args.Any(arg => string.Equals(arg, "--move-proof-presented-frame-smoke", StringComparison.OrdinalIgnoreCase))) @@ -1909,6 +1924,58 @@ static async Task RunMuninnMoveLiveRosterSmokeAsync(IReadOnlyList i : 1; } +static int RunMuninnMoveOdinLiveSmoke(string odinCultMeshUri) +{ + const string providerId = "muninn.telemetry.nightwing"; + const string streamId = "muninn:nightwing:move-evidence"; + var discovered = MimirOdinMoveProofEvidenceRingProvider.Discover(odinCultMeshUri, providerId, streamId); + var configuration = new MimirMoveProofRuntimeConfiguration + { + OdinCultMeshUri = odinCultMeshUri, + MuninnProviderId = providerId, + EvidenceStreamId = streamId + }; + if (!MimirOdinMoveProofEvidenceRingProvider.Instance.TryOpenEvidenceRing( + configuration, + out var lease, + out var diagnostic) || lease is null) + { + throw new InvalidOperationException(diagnostic); + } + + using (lease) + { + var deadline = DateTime.UtcNow + TimeSpan.FromSeconds(8); + while (DateTime.UtcNow < deadline) + { + if (lease.Ring.TryAcquireLatestRead(out var frameLease)) + { + using (frameLease) + { + var frame = MimirMuninnMoveEvidenceAdapter.DeserializeStreamFrame( + frameLease.Memory[..frameLease.Handle.ByteLength]); + var ok = frame.FrameId.StartsWith($"{streamId}:", StringComparison.Ordinal) && + frame.ControllerStates.Length >= 1 && + discovered.Channel == "move-evidence"; + Console.WriteLine(JsonSerializer.Serialize(new + { + ok, + diagnostic, + discovered, + frame.FrameId, + frame.ProducerPeerId, + controllers = frame.ControllerStates.Select(state => state.MoveId).Distinct().Order().ToArray(), + markers = frame.MarkerCandidates.Length + })); + return ok ? 0 : 1; + } + } + Thread.Sleep(20); + } + } + throw new TimeoutException($"No live frame arrived from Odin-discovered stream {streamId}."); +} + static int RunMuninnMoveCultMeshStreamSmoke(string nativeReservoirPath) { if (!File.Exists(nativeReservoirPath)) @@ -2308,7 +2375,8 @@ static int RunMoveProofRuntimeDriverSmoke( string nativeReservoirPath, bool activateViaConfiguredSource, bool requireActivationSurface = false, - bool activateViaSnapshotProvider = false) + bool activateViaSnapshotProvider = false, + string snapshotOutputPath = "") { if (!File.Exists(nativeReservoirPath)) { @@ -2320,8 +2388,11 @@ static int RunMoveProofRuntimeDriverSmoke( const ulong rightWitnessHash = 0x0000000000000B0B; const string observedAt = "2026-06-11T18:30:00.0040000Z"; var streamId = "muninn:nightwing:move-evidence"; + var deleteSnapshotOnExit = activateViaSnapshotProvider && string.IsNullOrWhiteSpace(snapshotOutputPath); var snapshotPath = activateViaSnapshotProvider - ? Path.Combine(Path.GetTempPath(), $"mimir-move-proof-evidence-{Guid.NewGuid():N}.mpack") + ? deleteSnapshotOnExit + ? Path.Combine(Path.GetTempPath(), $"mimir-move-proof-evidence-{Guid.NewGuid():N}.mpack") + : Path.GetFullPath(snapshotOutputPath) : ""; var frame = new MimirMuninnMoveEvidenceStreamFrame( FrameId: "muninn:nightwing:move-evidence:79", @@ -2555,7 +2626,7 @@ static int RunMoveProofRuntimeDriverSmoke( duplicateSuppressed ? 0 : 1; - if (!string.IsNullOrWhiteSpace(snapshotPath) && File.Exists(snapshotPath)) + if (deleteSnapshotOnExit && File.Exists(snapshotPath)) { File.Delete(snapshotPath); } @@ -2563,6 +2634,195 @@ static int RunMoveProofRuntimeDriverSmoke( return result; } +static int RunMoveProofRuntimeSnapshotFileSmoke( + string nativeReservoirPath, + string snapshotPath, + bool requireFullPose) +{ + if (!File.Exists(nativeReservoirPath)) + { + Console.Error.WriteLine($"move-proof-runtime-snapshot-file-smoke missing-native path={nativeReservoirPath}"); + return 1; + } + + if (string.IsNullOrWhiteSpace(snapshotPath) || !File.Exists(snapshotPath)) + { + Console.Error.WriteLine($"move-proof-runtime-snapshot-file-smoke missing-snapshot path={snapshotPath}"); + return 1; + } + + var absoluteSnapshotPath = Path.GetFullPath(snapshotPath); + MimirMoveProofEvidenceFrameSnapshotDocument snapshot; + MimirMuninnMoveEvidenceStreamFrame decoded; + try + { + snapshot = MimirMoveProofEvidenceFrameSnapshot.Load(absoluteSnapshotPath); + decoded = MimirMuninnMoveEvidenceAdapter.DeserializeStreamFrame(snapshot.Payload); + } + catch (Exception ex) when (ex is IOException or UnauthorizedAccessException or MessagePackSerializationException) + { + Console.Error.WriteLine($"move-proof-runtime-snapshot-file-smoke snapshot-decode-failed path={absoluteSnapshotPath} error={ex.Message}"); + return 1; + } + + var suffix = SequenceSuffix(decoded.FrameId); + var decodedSamples = MimirMuninnMoveEvidenceAdapter.BuildNativeSamples( + decoded.MarkerCandidates, + decoded.ControllerStates); + var markerSamples = decodedSamples + .Where(sample => sample.EvidenceKind == (uint)MimirNativeMoveEvidenceKind.OpticalMarker) + .ToArray(); + var controllerSamples = decodedSamples + .Where(sample => sample.EvidenceKind == (uint)MimirNativeMoveEvidenceKind.ControllerState) + .ToArray(); + var distinctMarkerWitnesses = markerSamples + .Select(sample => sample.WitnessIdHash) + .Where(hash => hash != 0UL) + .Distinct() + .Count(); + var markerWitnessSummary = string.Join( + ",", + decoded.MarkerCandidates + .GroupBy(marker => marker.SourceIdHash) + .OrderBy(group => group.Key) + .Select(group => + $"0x{group.Key:X16}:{group.Count()}@peak{group.Max(marker => marker.PeakLuma)}:area{group.Max(marker => marker.AreaPx)}:radius{group.Max(marker => marker.RadiusPx):0.0}:score{group.Max(marker => marker.Score):0.000}")); + var nearestControllerSkewMs = markerSamples.Length == 0 || controllerSamples.Length == 0 + ? double.NaN + : markerSamples + .Select(marker => controllerSamples + .Min(controller => TimestampDelta(marker.SourceTimestampNs, controller.SourceTimestampNs))) + .Min() / 1_000_000.0; + var configuration = CreateMoveProofConfigurationForSnapshot( + nativeReservoirPath, + absoluteSnapshotPath, + snapshot, + decoded); + var configErrors = configuration.Validate(); + if (configErrors.Length > 0) + { + Console.Error.WriteLine($"move-proof-runtime-snapshot-file-smoke config-invalid errors={string.Join("|", configErrors)}"); + return 1; + } + + using var runtime = new MimirRuntime( + new AquariumRuntimeOptions(Headless: true, CultCachePath: ""), + new MimirRuntimeConfiguration + { + Settings = new MimirSynchronizationSettings(), + MoveProofSources = [configuration], + }); + runtime.OnSceneReady(); + var input = new InputState(); + runtime.Update(0.016f, input); + var proofFrame = runtime.Frame; + var proofSurface = runtime.LatestMoveProofSurface; + var proofSplineCount = proofFrame.Scene.SplineFrame.Splines.Count(spline => + spline.Id.StartsWith("move-proof-", StringComparison.Ordinal)); + var activationStatus = runtime.MoveProofActivationStatuses.LastOrDefault(status => + string.Equals(status.EvidenceStreamId, configuration.EvidenceStreamId, StringComparison.Ordinal)); + var activationDocument = runtime + .CreateMoveProofRuntimeActivationDocuments(snapshot.CapturedAtNs == 0 ? (ulong)decoded.PublishedAtNs : snapshot.CapturedAtNs) + .SingleOrDefault(document => string.Equals(document.EvidenceStreamId, configuration.EvidenceStreamId, StringComparison.Ordinal)); + var expectedMimirEvidenceFrameId = $"{configuration.MimirEvidenceFramePrefix}:{suffix}"; + var expectedMimirPoseFrameId = $"{configuration.MimirPoseFramePrefix}:{suffix}"; + var expectedFensalirFrameId = $"{configuration.FensalirFramePrefix}:{suffix}"; + + Console.WriteLine( + $"move-proof-runtime-snapshot-file-smoke snapshot=\"{absoluteSnapshotPath}\" proof={proofSurface?.ProofId ?? "none"} chain={proofSurface?.MuninnEvidenceFrameId ?? "none"}->{proofSurface?.MimirEvidenceFrameId ?? "none"}->{proofSurface?.MimirPoseFrameId ?? "none"}->{proofSurface?.FensalirFrameId ?? "none"} verdict={proofSurface?.Verdict.ToString() ?? "none"} fullPose={proofSurface?.IsFullPose.ToString() ?? "False"} markers={proofSurface?.OpticalMarkerCount.ToString() ?? "0"} controllers={proofSurface?.ControllerStateCount.ToString() ?? "0"} poses={proofSurface?.PoseCount.ToString() ?? "0"} markerWitnesses={distinctMarkerWitnesses} markerWitnessSummary=\"{markerWitnessSummary}\" nearestControllerSkewMs={nearestControllerSkewMs:0.000} failure=\"{proofSurface?.FailureReason ?? ""}\" splines={proofSplineCount} activationActive={activationStatus?.Active.ToString() ?? "False"} activationDiagnostic=\"{activationStatus?.Diagnostic ?? ""}\" activationDocumentActive={activationDocument?.Active.ToString() ?? "False"} requireFullPose={requireFullPose}"); + + return proofSurface is not null && + activationStatus?.Active == true && + activationDocument is { Active: true, DriverRegistered: true } && + activationDocument.Diagnostic.Contains("one-copy evidence snapshot supplied", StringComparison.Ordinal) && + proofSurface.MuninnEvidenceFrameId == decoded.FrameId && + proofSurface.MimirEvidenceFrameId == expectedMimirEvidenceFrameId && + proofSurface.MimirPoseFrameId == expectedMimirPoseFrameId && + proofSurface.FensalirFrameId == expectedFensalirFrameId && + proofFrame.Scene.SplineFrame.HasInput && + proofSplineCount >= 1 && + (!requireFullPose || proofSurface.IsFullPose) + ? 0 + : 1; +} + +static MimirMoveProofRuntimeConfiguration CreateMoveProofConfigurationForSnapshot( + string nativeReservoirPath, + string snapshotPath, + MimirMoveProofEvidenceFrameSnapshotDocument snapshot, + MimirMuninnMoveEvidenceStreamFrame frame) +{ + var cameras = frame.MarkerCandidates + .Where(marker => marker.SourceIdHash != 0UL) + .GroupBy(marker => marker.SourceIdHash) + .Select((group, index) => + { + var marker = group.First(); + var side = index == 0 ? -0.1 : 0.1; + return new MimirMoveProofCameraCalibrationConfiguration + { + CameraId = string.IsNullOrWhiteSpace(marker.CameraId) + ? $"snapshot-camera-{index}" + : $"{marker.HostId}:{marker.CameraId}", + WitnessIdHash = group.Key, + PositionMeters = new MimirVector3Snapshot(side, 0.0, 0.0), + Orientation = new MimirQuaternionSnapshot(0.0, 0.0, 0.0, 1.0), + FocalLengthXPx = 100.0, + FocalLengthYPx = 100.0, + PrincipalPointXPx = 160.0, + PrincipalPointYPx = 120.0, + }; + }) + .Take(2) + .ToList(); + while (cameras.Count < 2) + { + var index = cameras.Count; + cameras.Add(new MimirMoveProofCameraCalibrationConfiguration + { + CameraId = $"snapshot-calibration-placeholder-{index}", + WitnessIdHash = 0xF00D_0000_0000_0001UL + (ulong)index, + PositionMeters = new MimirVector3Snapshot(index == 0 ? -0.1 : 0.1, 0.0, 0.0), + Orientation = new MimirQuaternionSnapshot(0.0, 0.0, 0.0, 1.0), + FocalLengthXPx = 100.0, + FocalLengthYPx = 100.0, + PrincipalPointXPx = 160.0, + PrincipalPointYPx = 120.0, + }); + } + + return new MimirMoveProofRuntimeConfiguration + { + EvidenceStreamId = snapshot.EvidenceStreamId, + EvidenceSnapshotPath = snapshotPath, + NativeReservoirPath = nativeReservoirPath, + MimirEvidenceSourceId = "mimir:starfire:move-evidence", + MimirEvidenceFramePrefix = "mimir:starfire:move-evidence", + MimirPoseFramePrefix = "mimir:starfire:move-pose", + MimirPoseProducerPeerId = "mimir:starfire", + FensalirFramePrefix = "fensalir:starfire:presented-frame", + FusionAuthorityId = "mimir.runtime.move-fusion", + ConsumerContract = "fensalir.move-controller-input", + Calibration = new MimirMoveProofCalibrationConfiguration + { + CalibrationId = "mimir-move-snapshot-replay-calibration-v1", + TrackingSpaceId = "mimir-stage-space", + Cameras = cameras, + MaximumAssociationSkewMilliseconds = 1000.0, + }, + }; +} + +static string SequenceSuffix(string frameId) +{ + var separator = frameId.LastIndexOf(':'); + return separator >= 0 && separator < frameId.Length - 1 + ? frameId[(separator + 1)..] + : frameId; +} + +static ulong TimestampDelta(ulong left, ulong right) => left >= right ? left - right : right - left; + static int RunMoveProofPresentedFrameSmoke(string outputPath) { var absoluteOutputPath = Path.GetFullPath(outputPath); diff --git a/src/Mimir.Runtime/Mimir.Runtime.csproj b/src/Mimir.Runtime/Mimir.Runtime.csproj index 84cbe2f..4d06a5e 100644 --- a/src/Mimir.Runtime/Mimir.Runtime.csproj +++ b/src/Mimir.Runtime/Mimir.Runtime.csproj @@ -14,6 +14,7 @@ + diff --git a/src/Mimir.Runtime/Synchronization/MimirMoveProofRuntimeActivation.cs b/src/Mimir.Runtime/Synchronization/MimirMoveProofRuntimeActivation.cs index 73d9bc5..969c21e 100644 --- a/src/Mimir.Runtime/Synchronization/MimirMoveProofRuntimeActivation.cs +++ b/src/Mimir.Runtime/Synchronization/MimirMoveProofRuntimeActivation.cs @@ -15,17 +15,23 @@ public interface IMimirMoveProofEvidenceRingProvider public sealed class MimirMoveProofEvidenceRingLease : IDisposable { private readonly bool ownsRing; + private readonly IDisposable? transportOwner; - public MimirMoveProofEvidenceRingLease(CultMeshSharedMemoryFrameRing ring, bool ownsRing = false) + public MimirMoveProofEvidenceRingLease( + CultMeshSharedMemoryFrameRing ring, + bool ownsRing = false, + IDisposable? transportOwner = null) { Ring = ring ?? throw new ArgumentNullException(nameof(ring)); this.ownsRing = ownsRing; + this.transportOwner = transportOwner; } public CultMeshSharedMemoryFrameRing Ring { get; } public void Dispose() { + transportOwner?.Dispose(); if (ownsRing) { Ring.Dispose(); @@ -42,7 +48,7 @@ public sealed record MimirMoveProofEvidenceFrameSnapshotDocument( [property: Key(1)] string EvidenceStreamId, [property: Key(2)] string FrameId, [property: Key(3)] string ProducerPeerId, - [property: Key(4)] long PublishedAtNs, + [property: Key(4)] ulong PublishedAtNs, [property: Key(5)] ulong CapturedAtNs, [property: Key(6)] byte[] Payload); @@ -65,7 +71,7 @@ public static class MimirMoveProofEvidenceFrameSnapshot EvidenceStreamId: StreamIdFromFrameId(frame.FrameId), FrameId: frame.FrameId, ProducerPeerId: frame.ProducerPeerId, - PublishedAtNs: frame.PublishedAtNs, + PublishedAtNs: checked((ulong)frame.PublishedAtNs), CapturedAtNs: capturedAtNs, Payload: payload); } @@ -192,6 +198,14 @@ public sealed class MimirConfiguredMoveProofEvidenceRingProvider : IMimirMovePro out string diagnostic) { ArgumentNullException.ThrowIfNull(configuration); + if (!string.IsNullOrWhiteSpace(configuration.OdinCultMeshUri)) + { + return MimirOdinMoveProofEvidenceRingProvider.Instance.TryOpenEvidenceRing( + configuration, + out lease, + out diagnostic); + } + if (string.IsNullOrWhiteSpace(configuration.EvidenceSnapshotPath)) { return MimirUnavailableMoveProofEvidenceRingProvider.Instance.TryOpenEvidenceRing( @@ -219,7 +233,7 @@ public sealed class MimirConfiguredMoveProofEvidenceRingProvider : IMimirMovePro var decoded = MimirMuninnMoveEvidenceAdapter.DeserializeStreamFrame(snapshot.Payload); if (!string.Equals(decoded.FrameId, snapshot.FrameId, StringComparison.Ordinal) || !string.Equals(decoded.ProducerPeerId, snapshot.ProducerPeerId, StringComparison.Ordinal) || - decoded.PublishedAtNs != snapshot.PublishedAtNs) + decoded.PublishedAtNs != checked((long)snapshot.PublishedAtNs)) { lease = null; diagnostic = "Move proof evidence snapshot metadata does not match its encoded Muninn frame payload."; @@ -230,7 +244,7 @@ public sealed class MimirConfiguredMoveProofEvidenceRingProvider : IMimirMovePro configuration.EvidenceStreamId, slotCount: 1, slotByteLength: snapshot.Payload.Length); - if (!ring.TryPublishCopy(snapshot.Payload, snapshot.PublishedAtNs, durationNs: 0, out _)) + if (!ring.TryPublishCopy(snapshot.Payload, checked((long)snapshot.PublishedAtNs), durationNs: 0, out _)) { ring.Dispose(); lease = null; diff --git a/src/Mimir.Runtime/Synchronization/MimirMoveProofRuntimeConfiguration.cs b/src/Mimir.Runtime/Synchronization/MimirMoveProofRuntimeConfiguration.cs index 6971a2a..54dabf4 100644 --- a/src/Mimir.Runtime/Synchronization/MimirMoveProofRuntimeConfiguration.cs +++ b/src/Mimir.Runtime/Synchronization/MimirMoveProofRuntimeConfiguration.cs @@ -10,6 +10,10 @@ public sealed class MimirMoveProofRuntimeConfiguration public string EvidenceSnapshotPath { get; init; } = ""; + public string OdinCultMeshUri { get; init; } = ""; + + public string MuninnProviderId { get; init; } = "muninn.telemetry.nightwing"; + public string NativeReservoirPath { get; init; } = ""; public string MimirEvidenceSourceId { get; init; } = "mimir:starfire:move-evidence"; @@ -37,6 +41,10 @@ public sealed class MimirMoveProofRuntimeConfiguration var errors = new List(); Require(EvidenceStreamId, nameof(EvidenceStreamId), errors); + if (!string.IsNullOrWhiteSpace(OdinCultMeshUri)) + { + Require(MuninnProviderId, nameof(MuninnProviderId), errors); + } Require(NativeReservoirPath, nameof(NativeReservoirPath), errors); Require(MimirEvidenceSourceId, nameof(MimirEvidenceSourceId), errors); Require(MimirEvidenceFramePrefix, nameof(MimirEvidenceFramePrefix), errors); diff --git a/src/Mimir.Runtime/Synchronization/MimirOdinMoveProofEvidenceRingProvider.cs b/src/Mimir.Runtime/Synchronization/MimirOdinMoveProofEvidenceRingProvider.cs new file mode 100644 index 0000000..26faafb --- /dev/null +++ b/src/Mimir.Runtime/Synchronization/MimirOdinMoveProofEvidenceRingProvider.cs @@ -0,0 +1,216 @@ +using System.Net; +using System.Net.Sockets; +using System.Text; +using System.Text.Json; +using GameCult.Mesh; +using GameCult.Networking; +using MessagePack; + +namespace Mimir.Runtime.Synchronization; + +public sealed record MimirDiscoveredCultMeshInputStream( + string ProviderId, + string StreamId, + string Schema, + string Address, + string Channel, + uint ConnectionId); + +public sealed class MimirOdinMoveProofEvidenceRingProvider : IMimirMoveProofEvidenceRingProvider +{ + private const uint OdinSnapshotConnectionId = 0x0d1d0002; + public static MimirOdinMoveProofEvidenceRingProvider Instance { get; } = new(); + + private MimirOdinMoveProofEvidenceRingProvider() { } + + public bool TryOpenEvidenceRing( + MimirMoveProofRuntimeConfiguration configuration, + out MimirMoveProofEvidenceRingLease? lease, + out string diagnostic) + { + ArgumentNullException.ThrowIfNull(configuration); + lease = null; + MimirMoveEvidenceRudpPump? pump = null; + CultMeshSharedMemoryFrameRing? ring = null; + try + { + var stream = Discover(configuration.OdinCultMeshUri, configuration.MuninnProviderId, configuration.EvidenceStreamId); + ring = new CultMeshSharedMemoryFrameRing(configuration.EvidenceStreamId, slotCount: 4, slotByteLength: 256 * 1024); + pump = new MimirMoveEvidenceRudpPump(stream, ring); + pump.Start(); + lease = new MimirMoveProofEvidenceRingLease(ring, ownsRing: true, transportOwner: pump); + diagnostic = $"Odin discovered {stream.ProviderId}/{stream.StreamId}; lowering {stream.Address}/{stream.Channel} into local CultMesh ring"; + return true; + } + catch (Exception error) + { + pump?.Dispose(); + ring?.Dispose(); + diagnostic = error.Message; + return false; + } + } + + public static MimirDiscoveredCultMeshInputStream Discover(string odinCultMeshUri, string providerId, string streamId) + { + ArgumentException.ThrowIfNullOrWhiteSpace(odinCultMeshUri); + ArgumentException.ThrowIfNullOrWhiteSpace(providerId); + ArgumentException.ThrowIfNullOrWhiteSpace(streamId); + var endpoint = CultMesh.ResolveRudpEndpoint(odinCultMeshUri); + using var transport = CultMesh.CreateRudpClient("mimir-odin-discovery", OdinSnapshotConnectionId, endpoint); + if (!transport.ConnectAndWait([], TimeSpan.FromSeconds(5), TimeSpan.FromMilliseconds(10))) + { + throw new TimeoutException($"Timed out connecting to Odin at {odinCultMeshUri}."); + } + + var messageId = $"mimir-muninn-discovery:{Guid.NewGuid():N}"; + transport.SendSchemaMessage(new CultNetSnapshotRequestMessage + { + MessageId = messageId, + SchemaIds = ["gamecult.eve.surface_state.v1"], + RecordKeys = ["surface:gamecult.network.status"] + }); + var deadline = DateTime.UtcNow + TimeSpan.FromSeconds(5); + while (DateTime.UtcNow < deadline) + { + var message = transport.ReceiveSchemaMessage(TimeSpan.FromMilliseconds(250), TimeSpan.FromMilliseconds(10)); + if (message is CultNetErrorMessage error) + { + throw new InvalidOperationException(error.Error); + } + if (message is CultNetSnapshotResponseRawMessage response && + string.Equals(response.MessageId, messageId, StringComparison.Ordinal)) + { + return SelectStream(response, providerId, streamId); + } + } + throw new TimeoutException($"Timed out waiting for Odin provider snapshot from {odinCultMeshUri}."); + } + + public static MimirDiscoveredCultMeshInputStream SelectStream( + CultNetSnapshotResponseRawMessage response, + string providerId, + string streamId) + { + var document = response.Documents.FirstOrDefault(candidate => + string.Equals(candidate.SchemaId, "gamecult.eve.surface_state.v1", StringComparison.Ordinal) && + string.Equals(candidate.RecordKey, "surface:gamecult.network.status", StringComparison.Ordinal)) + ?? throw new InvalidOperationException("Odin snapshot omitted surface:gamecult.network.status."); + using var json = JsonDocument.Parse(MessagePackSerializer.ConvertToJson(document.Payload)); + if (!json.RootElement.TryGetProperty("providerCatalog", out var providers)) + { + throw new InvalidOperationException("Odin network status omitted providerCatalog."); + } + foreach (var provider in providers.EnumerateArray()) + { + if (!string.Equals(Text(provider, "id"), providerId, StringComparison.Ordinal) || + !provider.TryGetProperty("inputStreams", out var streams)) + { + continue; + } + foreach (var stream in streams.EnumerateArray()) + { + if (!string.Equals(Text(stream, "streamId"), streamId, StringComparison.Ordinal) || + !string.Equals(Text(stream, "schema"), MimirMuninnMoveEvidenceAdapter.StreamMetadataSchemaId, StringComparison.Ordinal) || + !string.Equals(Text(stream, "transport"), "cultnet.transport.rudp.v0", StringComparison.Ordinal)) + { + continue; + } + return new MimirDiscoveredCultMeshInputStream( + providerId, + streamId, + Text(stream, "schema"), + Text(stream, "address"), + Text(stream, "channel"), + stream.GetProperty("connectionId").GetUInt32()); + } + throw new InvalidOperationException($"Odin provider '{providerId}' has no compatible input stream '{streamId}'."); + } + throw new InvalidOperationException($"Odin has no provider '{providerId}'."); + } + + private static string Text(JsonElement value, string name) => + value.TryGetProperty(name, out var property) ? property.GetString() ?? "" : ""; +} + +internal sealed class MimirMoveEvidenceRudpPump : IDisposable +{ + private readonly MimirDiscoveredCultMeshInputStream stream; + private readonly CultMeshSharedMemoryFrameRing ring; + private readonly CancellationTokenSource stopping = new(); + private readonly ManualResetEventSlim started = new(false); + private Thread? thread; + private Exception? startupError; + + public MimirMoveEvidenceRudpPump(MimirDiscoveredCultMeshInputStream stream, CultMeshSharedMemoryFrameRing ring) + { + this.stream = stream; + this.ring = ring; + } + + public void Start() + { + thread = new Thread(Run) { IsBackground = true, Name = "mimir-muninn-move-evidence" }; + thread.Start(); + if (!started.Wait(TimeSpan.FromSeconds(6))) + { + throw new TimeoutException($"Timed out starting Move evidence receiver for {stream.Address}."); + } + if (startupError is not null) + { + throw new InvalidOperationException( + $"Could not connect Odin-discovered Move evidence stream {stream.StreamId} at {stream.Address}: {startupError.Message}", + startupError); + } + } + + private void Run() + { + try + { + using var transport = CultMesh.CreateRudpClient( + "mimir-move-evidence-consumer", + stream.ConnectionId, + CultMesh.ResolveRudpEndpoint($"rudp://{stream.Address}")); + if (!transport.ConnectAndWait([], TimeSpan.FromSeconds(5), TimeSpan.FromMilliseconds(5))) + { + throw new TimeoutException($"Timed out connecting to {stream.Address}."); + } + transport.Send("hid.subscribe", Encoding.UTF8.GetBytes(JsonSerializer.Serialize(new { streamId = stream.StreamId }))); + started.Set(); + while (!stopping.IsCancellationRequested) + { + transport.PollResends(); + var frame = transport.ReceiveOnce(); + if (frame is null || !string.Equals(frame.ChannelId, stream.Channel, StringComparison.Ordinal)) + { + Thread.Sleep(2); + continue; + } + var decoded = MimirMuninnMoveEvidenceAdapter.DeserializeStreamFrame(frame.Payload); + if (!string.Equals(decoded.FrameId, stream.StreamId, StringComparison.Ordinal) && + !decoded.FrameId.StartsWith($"{stream.StreamId}:", StringComparison.Ordinal)) + { + continue; + } + ring.TryPublishCopy(frame.Payload, decoded.PublishedAtNs, durationNs: 0, out _); + } + } + catch (Exception error) + { + if (!started.IsSet) + { + startupError = error; + started.Set(); + } + } + } + + public void Dispose() + { + stopping.Cancel(); + thread?.Join(TimeSpan.FromSeconds(2)); + started.Dispose(); + stopping.Dispose(); + } +}