Migrate Mimir media bridge to RUDP

This commit is contained in:
Meta Vens 2026-06-16 18:08:40 +01:00
commit 4ed4ba190b
13 changed files with 771 additions and 36 deletions

View file

@ -6,5 +6,6 @@
<Project Path="src/Mimir.EveSensorReceiver/Mimir.EveSensorReceiver.csproj" />
<Project Path="src/Mimir.VerseRecorder/Mimir.VerseRecorder.csproj" />
<Project Path="src/Mimir.PsMoveProbe/Mimir.PsMoveProbe.csproj" />
<Project Path="src/Mimir.CultMeshMedia/Mimir.CultMeshMedia.csproj" />
</Folder>
</Solution>

View file

@ -279,14 +279,14 @@ a named invariant that the native runtime cannot protect yet.
controls/previews/stats, Yggdrasil publishes to the site, and OBS is
compatibility-only.
- `src/Mimir.CultMeshMedia` is the first explicit CultMesh media/body bridge:
`relay` hosts the document relay on CultNet UDP `3075`, `send` reads an
`relay` hosts the document relay on CultNet RUDP `3075`, `send` reads an
MPEG-TS byte stream from stdin and publishes rolling
`mimir.cultmesh_media_frame` documents, and `recv` subscribes to those
documents and writes ordered MPEG-TS bytes to a Starfire-local UDP endpoint
for compatibility sinks. Source audit: its C# implementation still uses older
`CultMesh.StartNodeAsync`/`CultMesh.ConnectClient` entrypoints; migrate the
bridge to CultLib's explicit C# RUDP helpers before treating it as fully
aligned with the daemon-swarm transport cut.
for compatibility sinks. Its C# implementation now uses explicit CultLib
RUDP client/session helpers; the older
`CultMesh.StartNodeAsync`/`CultMesh.ConnectClient` entrypoints no longer own
the media transport lane.
- `scripts/start-raven-cultmesh-av-sender.ps1`,
`scripts/start-yggdrasil-cultmesh-media-relay.ps1`, and
`scripts/start-starfire-cultmesh-av-receiver.ps1` are the CultMesh bridge
@ -322,9 +322,9 @@ a named invariant that the native runtime cannot protect yet.
discovery, Idunn owns keepalive decisions, and Mimir-owned dashboard/reference
surfaces only report their own observed state. TCP/HTTP/WebSocket stay
lowerings or compatibility evidence.
- `Mimir.CultMeshMedia` still needs an explicit RUDP transport cut. Do not add
another TCP bridge, HTTP status shim, or WebSocket-derived service truth while
the C# RUDP helpers exist.
- `Mimir.CultMeshMedia` has completed its explicit RUDP transport cut. Do not
add another TCP bridge, HTTP status shim, or WebSocket-derived service truth
while the RUDP document lane exists.
- Process-backed stream sources are only acceptable for network bridge feeds or
diagnostics. Six-camera local ingest belongs behind direct capture drivers.
- Frame-event process sources are diagnostic only. They prove source cadence and
@ -337,9 +337,10 @@ a named invariant that the native runtime cannot protect yet.
1. Replace the frame-event diagnostic bridge with concrete direct capture
drivers for Leap stereo IR first, then the
other cameras.
2. Cut `src/Mimir.CultMeshMedia` from older CultMesh node/client entrypoints to
explicit CultNet RUDP helpers, preferably authorized-peer dialing where the
peer catalog exists. Preserve OBS-local UDP as egress only.
2. Cut the remaining dashboard, recorder, and witness service truth paths from
older TCP/WebSocket-derived assumptions to explicit CultNet RUDP records,
preferably authorized-peer dialing where the peer catalog exists. Preserve
OBS-local UDP as egress only for compatibility sinks.
3. Feed those drivers into `MimirVideoCaptureDriverSource` and prove sustained
frame cadence in the rolling buffers.
4. Promote the packet-song physical calibration receipt into the runtime

View file

@ -257,13 +257,12 @@ video archive. Yggdrasil relays CultNet reliable UDP over the WireGuard mesh.
Starfire lowers the subscribed stream to local UDP for compatibility sinks
because OBS is not a CultMesh runtime.
Source audit correction: the desired transport is explicit CultNet RUDP, and
CultLib now provides C# `CultMesh.CreateRudpServer`,
`CultMesh.CreateRudpClient`, and authorized-peer helper APIs. The current
`Mimir.CultMeshMedia` source still uses older `CultMesh.StartNodeAsync` and
`CultMesh.ConnectClient` entrypoints. Treat that as migration debt: the bridge
should be cut to explicit RUDP helpers before it is considered aligned with the
daemon-swarm transport doctrine.
Source audit correction resolved: `Mimir.CultMeshMedia` now uses explicit
CultLib RUDP helpers for the sender/receiver and a relay-owned RUDP
socket/session loop for multi-peer forwarding. The older
`CultMesh.StartNodeAsync` and `CultMesh.ConnectClient` path no longer owns this
media lane. The relay still persists received puts to CultCache as evidence and
for inspection; live network transit is the RUDP schema-document path.
Current deployment: Yggdrasil runs the relay from
`/opt/gamecult/mimir-cultmesh-media/Mimir.CultMeshMedia` with cache
@ -315,7 +314,8 @@ browser/dashboard WebSocket paths are still renderer/client lowerings. They may
remain as lowering adapters, but any provider catalog, retained service state,
command boundary, transport profile, or lifecycle health claim must be
published as typed CultNet/RUDP records instead of being inferred from those
TCP/WebSocket surfaces.
TCP/WebSocket surfaces. The media-frame bridge is already on that side of the
line.
## Audio Field

View file

@ -25,10 +25,10 @@ Get-Content .\state\evidence.jsonl -Tail 8
- V1 now also has an explicit CultMesh media bridge for Raven program feeds:
`src/Mimir.CultMeshMedia` sends rolling `mimir.cultmesh_media_frame`
documents over CultNet reliable UDP through Yggdrasil and lowers them on
Starfire to local MPEG-TS UDP for OBS. Source audit warning: the current C#
bridge still uses older `CultMesh.StartNodeAsync`/`CultMesh.ConnectClient`;
migrate it to explicit C# RUDP helpers before treating the media bridge as
fully aligned with the swarm transport cut.
Starfire to local MPEG-TS UDP for OBS. The bridge now uses explicit CultLib
RUDP helpers/session plumbing; the old
`CultMesh.StartNodeAsync`/`CultMesh.ConnectClient` path no longer owns this
media lane.
- CultLib has moved the daemon-swarm direction to CultNet over RUDP across
runtimes. Mimir's typed daemon truth should default to
`cultnet.transport.rudp.v0`; TCP/HTTP/WebSocket are client lowerings,
@ -103,9 +103,9 @@ Get-Content .\state\evidence.jsonl -Tail 8
dashboard boundary cut is provider advertisement, retained state,
command_boundary, and transport_profile over RUDP so HTTP/WebSocket catalog
ingestion can be demoted to client/debug lowering.
- Cut `Mimir.CultMeshMedia` to explicit CultNet RUDP helpers next; no new TCP,
HTTP, or WebSocket surface should be allowed to become daemon or transport
truth.
- Cut the remaining Eve dashboard, recorder, and witness service truth paths to
explicit CultNet RUDP records next; no new TCP, HTTP, or WebSocket surface
should be allowed to become daemon or transport truth.
- Kiyo Pro has two UVC extension units. Moving it to a motherboard USB3 port
changed the descriptor path to `root_hub30` / `bcdUSB=0x0320` and exposed
720p/1080p YUY2/MJPG/H264/NV12 at 60 fps, but Windows still reports

View file

@ -0,0 +1,89 @@
param(
[string]$RelayHost = "10.77.0.1",
[int]$RelayPort = 3075,
[string]$Stream = "raven-primary-av",
[string]$FfmpegPath = "ffmpeg",
[string]$VideoInput = "desktop",
[ValidateSet("WasapiLoopback", "DirectShow")]
[string]$AudioMode = "WasapiLoopback",
[string]$AudioInput = "Stereo Mix (Realtek(R) Audio)",
[ValidateSet("Console", "Multimedia", "Communications")]
[string]$LoopbackRole = "Console",
[int]$Width = 1920,
[int]$Height = 1080,
[int]$Framerate = 30,
[string]$VideoBitrate = "8000k",
[string]$AudioBitrate = "192k",
[string]$MediaToolPath = ""
)
$ErrorActionPreference = "Stop"
$project = Join-Path $PSScriptRoot "..\src\Mimir.CultMeshMedia\Mimir.CultMeshMedia.csproj"
$publishedTool = Join-Path $PSScriptRoot "..\artifacts\mimir-cultmesh-media-win-x64-selfcontained\Mimir.CultMeshMedia.exe"
$loopbackScript = Join-Path $PSScriptRoot "wasapi-loopback-capture.ps1"
if ([string]::IsNullOrWhiteSpace($MediaToolPath) -and (Test-Path $publishedTool)) {
$MediaToolPath = $publishedTool
}
$ffmpegArgs = @(
"-hide_banner",
"-loglevel", "warning",
"-f", "gdigrab",
"-framerate", $Framerate,
"-video_size", "${Width}x${Height}",
"-i", $VideoInput
)
if ($AudioMode -eq "WasapiLoopback") {
$ffmpegArgs += @(
"-f", "f32le",
"-ar", "48000",
"-ac", "2",
"-i", "pipe:0"
)
} else {
$ffmpegArgs += @(
"-f", "dshow",
"-i", "audio=$AudioInput"
)
}
$ffmpegArgs += @(
"-map", "0:v:0",
"-map", "1:a:0",
"-c:v", "h264_nvenc",
"-preset", "p3",
"-tune", "ll",
"-b:v", $VideoBitrate,
"-maxrate", $VideoBitrate,
"-bufsize", "16000k",
"-g", ($Framerate * 2),
"-pix_fmt", "yuv420p",
"-c:a", "aac",
"-b:a", $AudioBitrate,
"-ar", "48000",
"-ac", "2",
"-f", "mpegts",
"pipe:1"
)
$senderArgs = @("send", "--host", $RelayHost, "--port", $RelayPort, "--stream", $Stream, "--producer", "raven", "--chunk-bytes", "16384", "--slots", "96")
if ($AudioMode -eq "WasapiLoopback") {
if (-not (Test-Path $loopbackScript)) {
throw "WASAPI loopback script not found: $loopbackScript"
}
if ([string]::IsNullOrWhiteSpace($MediaToolPath)) {
& $loopbackScript -Output stdout -Role $LoopbackRole | & $FfmpegPath @ffmpegArgs | dotnet run --project $project -- @senderArgs
} else {
& $loopbackScript -Output stdout -Role $LoopbackRole | & $FfmpegPath @ffmpegArgs | & $MediaToolPath @senderArgs
}
} else {
if ([string]::IsNullOrWhiteSpace($MediaToolPath)) {
& $FfmpegPath @ffmpegArgs | dotnet run --project $project -- @senderArgs
} else {
& $FfmpegPath @ffmpegArgs | & $MediaToolPath @senderArgs
}
}

View file

@ -0,0 +1,10 @@
param(
[string]$RelayHost = "10.77.0.1",
[int]$RelayPort = 3075,
[string]$Stream = "raven-primary-av",
[string]$Udp = "127.0.0.1:5200"
)
$ErrorActionPreference = "Stop"
$project = Join-Path $PSScriptRoot "..\src\Mimir.CultMeshMedia\Mimir.CultMeshMedia.csproj"
dotnet run --project $project -- recv --host $RelayHost --port $RelayPort --stream $Stream --udp $Udp

View file

@ -0,0 +1,8 @@
param(
[string]$Cache = "C:\Meta\Mimir\state\mimir-cultmesh-media.cc"
)
$ErrorActionPreference = "Stop"
$project = Join-Path $PSScriptRoot "..\src\Mimir.CultMeshMedia\Mimir.CultMeshMedia.csproj"
New-Item -ItemType Directory -Force -Path (Split-Path -Parent $Cache) | Out-Null
dotnet run --project $project -- relay --cache $Cache

View file

@ -0,0 +1,18 @@
<Project Sdk="Microsoft.NET.Sdk">
<PropertyGroup>
<OutputType>Exe</OutputType>
<TargetFramework>net10.0</TargetFramework>
<AssemblyName>Mimir.CultMeshMedia</AssemblyName>
<RootNamespace>Mimir.CultMeshMedia</RootNamespace>
<ImplicitUsings>enable</ImplicitUsings>
<Nullable>enable</Nullable>
</PropertyGroup>
<ItemGroup>
<ProjectReference Include="..\Mimir.Runtime\Mimir.Runtime.csproj" />
<ProjectReference Include="..\..\..\CultLib\src\GameCult.Mesh\GameCult.Mesh.csproj" />
<ProjectReference Include="..\..\..\CultLib\src\GameCult.Networking\GameCult.Networking.csproj" />
</ItemGroup>
</Project>

View file

@ -0,0 +1,568 @@
using System.Diagnostics;
using System.Globalization;
using System.Net;
using System.Net.Sockets;
using GameCult.Caching;
using GameCult.Caching.MessagePack;
using GameCult.Mesh;
using GameCult.Networking;
using Mimir.Runtime.Synchronization;
const string SchemaName = "mimir.cultmesh_media_frame";
const string DefaultStreamId = "raven-primary-av";
const uint DefaultConnectionId = 0x4d4d_0101;
const int DefaultPort = 3075;
const int DefaultMaxFragmentBytes = 1200;
const int DefaultMaxPendingReliablePackets = 8192;
var command = args.Length > 0 ? args[0].Trim().ToLowerInvariant() : string.Empty;
var options = CliOptions.Parse(args.Skip(1));
return command switch
{
"relay" => await RunRelayAsync(options),
"send" => await RunSenderAsync(options),
"recv" or "receive" => await RunReceiverAsync(options),
_ => Usage()
};
static int Usage()
{
Console.Error.WriteLine("""
Usage:
Mimir.CultMeshMedia relay --cache <path> [--bind 0.0.0.0:3075]
Mimir.CultMeshMedia send --host <rudp-host> [--port 3075] [--stream raven-primary-av] [--chunk-bytes 16384] [--slots 96]
Mimir.CultMeshMedia recv --host <rudp-host> [--port 3075] [--stream raven-primary-av] [--udp 127.0.0.1:5200]
send reads an MPEG-TS byte stream from stdin and publishes rolling CultMesh media-frame documents over CultNet RUDP.
recv subscribes to those RUDP document changes and writes ordered payload bytes to a local UDP MPEG-TS endpoint for OBS.
""");
return 2;
}
static async Task<int> RunRelayAsync(CliOptions options)
{
var cachePath = options.Get("cache") ?? Path.Combine(Environment.CurrentDirectory, "mimir-cultmesh-media.cc");
var bind = ParseEndpoint(options.Get("bind") ?? $"0.0.0.0:{options.GetInt("port", DefaultPort)}");
var registry = CreateDocumentRegistry();
var schemaId = ResolveMediaSchemaId(registry);
using var cache = await CultCacheMessagePack.OpenAsync(cachePath).ConfigureAwait(false);
using var relay = new RudpMediaRelay(
bind,
cache,
registry,
schemaId,
options.GetUInt("connection-id", DefaultConnectionId),
options.GetInt("max-fragment-bytes", DefaultMaxFragmentBytes),
options.GetInt("max-pending-reliable", DefaultMaxPendingReliablePackets));
Console.WriteLine($"CultMesh media relay listening on rudp://{bind.Address}:{bind.Port}; cache={Path.GetFullPath(cachePath)}");
await relay.RunAsync().ConfigureAwait(false);
await cache.FlushAsync().ConfigureAwait(false);
return 0;
}
static async Task<int> RunSenderAsync(CliOptions options)
{
var endpoint = ResolveRudpEndpoint(options);
var streamId = options.Get("stream") ?? DefaultStreamId;
var chunkBytes = Math.Clamp(options.GetInt("chunk-bytes", 16 * 1024), 1316, 60 * 1024);
var slots = Math.Clamp(options.GetInt("slots", 96), 8, 4096);
var producer = options.Get("producer") ?? Environment.MachineName.ToLowerInvariant();
var registry = CreateDocumentRegistry();
using var transport = CreateRudpClient(
options.Get("runtime") ?? $"{producer}-mimir-media-sender",
endpoint,
options);
ConnectOrThrow(transport, endpoint.Uri);
Console.Error.WriteLine($"Sending stdin MPEG-TS to {endpoint.Uri}/{streamId}; chunkBytes={chunkBytes}; slots={slots}");
var input = Console.OpenStandardInput();
var buffer = new byte[chunkBytes];
long sequence = 0;
var started = Stopwatch.StartNew();
while (true)
{
var read = await input.ReadAsync(buffer.AsMemory(0, buffer.Length)).ConfigureAwait(false);
if (read <= 0)
{
break;
}
var payload = buffer.AsSpan(0, read).ToArray();
var document = new MimirCultMeshMediaFrameDocument(
FrameId: $"{streamId}:{sequence:D20}",
StreamId: streamId,
Sequence: sequence,
ProducedAtUtc: DateTimeOffset.UtcNow.ToString("O", CultureInfo.InvariantCulture),
TimestampNanoseconds: started.ElapsedTicks * 1_000_000_000L / Stopwatch.Frequency,
PayloadKind: "mpegts-chunk",
Container: "mpegts",
VideoCodec: options.Get("video-codec") ?? "h264",
AudioCodec: options.Get("audio-codec") ?? "aac",
PayloadBytes: payload.Length,
Payload: payload,
ProducerNode: producer,
ClockDomainId: options.Get("clock-domain") ?? "raven-monotonic",
Tags: ["raven", "program", "obs", "cultmesh-media", "cultnet.transport.rudp.v0"]);
var handle = new CultRecordHandle<MimirCultMeshMediaFrameDocument>(
new CultRecordKey($"{streamId}:slot:{sequence % slots:D4}"));
var message = registry.CreateRawDocumentPutMessage(
Guid.NewGuid().ToString("N"),
handle,
document,
new CultNetDocumentMessageOptions
{
SourceRuntimeId = producer,
SourceRole = "mimir-cultmesh-media-sender",
Tags = ["mimir", "media", streamId, "cultnet.transport.rudp.v0"]
});
transport.SendSchema(CultNetSchemaMessageSerialization.Serialize(message));
PumpClient(transport, TimeSpan.FromMilliseconds(5));
sequence++;
if (sequence % 120 == 0)
{
Console.Error.WriteLine($"sent sequence={sequence - 1}");
}
}
Console.Error.WriteLine($"sent chunks={sequence}; draining");
PumpClient(transport, TimeSpan.FromMilliseconds(options.GetInt("drain-ms", 1000)));
return 0;
}
static async Task<int> RunReceiverAsync(CliOptions options)
{
var endpoint = ResolveRudpEndpoint(options);
var streamId = options.Get("stream") ?? DefaultStreamId;
var udp = options.Get("udp") ?? "127.0.0.1:5200";
var udpEndpoint = ParseEndpoint(udp);
var registry = CreateDocumentRegistry();
var schemaId = ResolveMediaSchemaId(registry);
var lastSequence = -1L;
long frames = 0;
long bytes = 0;
var loggedFirstFrame = false;
using var socket = new UdpClient();
using var transport = CreateRudpClient(
options.Get("runtime") ?? $"{Environment.MachineName.ToLowerInvariant()}-mimir-media-receiver",
endpoint,
options);
ConnectOrThrow(transport, endpoint.Uri);
transport.SendSchema(CultNetSchemaMessageSerialization.Serialize(new CultNetDatabaseSubscribeMessage
{
MessageId = Guid.NewGuid().ToString("N"),
SubscriptionId = $"mimir-media-{streamId}",
SchemaIds = [schemaId],
IncludeSnapshot = false
}));
Console.WriteLine($"Receiving {endpoint.Uri}/{streamId}; writing MPEG-TS UDP to udp://{udp}");
while (true)
{
transport.PollResends();
if (transport.ReceiveOnce() is not { } frame)
{
await Task.Delay(2).ConfigureAwait(false);
continue;
}
if (!string.Equals(frame.ChannelId, "schema", StringComparison.Ordinal))
{
continue;
}
if (CultNetSchemaMessageSerialization.Deserialize(frame.Payload) is not CultNetDatabaseChangeRawMessage change ||
change.Document is not { } rawDocument ||
!string.Equals(rawDocument.SchemaId, schemaId, StringComparison.Ordinal))
{
continue;
}
var document = (MimirCultMeshMediaFrameDocument)registry
.GetBySchemaId(schemaId)!
.PayloadDeserializer(rawDocument.Payload);
if (!string.Equals(document.StreamId, streamId, StringComparison.Ordinal) ||
document.Sequence <= lastSequence)
{
continue;
}
lastSequence = document.Sequence;
socket.Send(document.Payload, document.Payload.Length, udpEndpoint);
frames++;
bytes += document.Payload.Length;
if (!loggedFirstFrame)
{
loggedFirstFrame = true;
Console.WriteLine($"received first sequence={document.Sequence} bytes={document.PayloadBytes}");
}
if (frames % 120 == 0)
{
Console.WriteLine($"received sequence={document.Sequence} frames={frames} bytes={bytes}");
}
}
}
static CultMeshRudpEndpoint ResolveRudpEndpoint(CliOptions options)
{
if (options.Get("endpoint") is { } endpoint)
{
return CultMesh.ParseRudpEndpoint(endpoint);
}
var host = options.Get("host") ?? "10.77.0.1";
var port = options.GetInt("port", DefaultPort);
return CultMesh.ParseRudpEndpoint($"rudp://{host}:{port}");
}
static CultNetRudpSocketTransportConnection CreateRudpClient(
string runtimeId,
CultMeshRudpEndpoint endpoint,
CliOptions options)
{
return CultMesh.CreateRudpClient(
runtimeId,
options.GetUInt("connection-id", DefaultConnectionId),
endpoint,
new CultMeshRudpSocketOptions
{
BindHost = options.Get("bind-host") ?? "0.0.0.0",
BindPort = options.GetInt("bind-port", 0),
ResendDelayMs = options.GetInt("resend-ms", 100),
MaxFragmentBytes = options.GetInt("max-fragment-bytes", DefaultMaxFragmentBytes),
MaxPendingReliablePackets = options.GetInt("max-pending-reliable", DefaultMaxPendingReliablePackets)
});
}
static void ConnectOrThrow(CultNetRudpSocketTransportConnection transport, string endpoint)
{
if (!transport.ConnectAndWait(Array.Empty<byte>(), TimeSpan.FromSeconds(10), TimeSpan.FromMilliseconds(5)))
{
throw new TimeoutException($"timed out connecting RUDP media transport to {endpoint}");
}
}
static void PumpClient(CultNetRudpSocketTransportConnection transport, TimeSpan duration)
{
var until = Stopwatch.StartNew();
while (until.Elapsed < duration)
{
transport.PollResends();
transport.ReceiveOnce();
Thread.Sleep(1);
}
}
static string ResolveMediaSchemaId(CultNetDocumentRegistry registry)
{
var binding = registry.GetByDocumentType(typeof(MimirCultMeshMediaFrameDocument)) ??
throw new InvalidOperationException($"CultNet document binding missing for {SchemaName}.");
return binding.SchemaId;
}
static CultNetDocumentRegistry CreateDocumentRegistry()
{
var documents = CultDocumentRegistry.Shared;
documents.GetRequired<MimirCultMeshMediaFrameDocument>();
return new CultNetDocumentRegistry(documents)
.Register(CultNetDocumentBinding.ForDocument<MimirCultMeshMediaFrameDocument>(documents));
}
static IPEndPoint ParseEndpoint(string value)
{
var split = value.LastIndexOf(':');
if (split <= 0 || split == value.Length - 1)
{
throw new ArgumentException($"Endpoint must be host:port, not '{value}'.");
}
var host = value[..split];
var port = int.Parse(value[(split + 1)..], CultureInfo.InvariantCulture);
var addresses = Dns.GetHostAddresses(host);
return new IPEndPoint(addresses.First(address => address.AddressFamily == AddressFamily.InterNetwork), port);
}
sealed class RudpMediaRelay : IDisposable
{
private readonly Socket socket;
private readonly CultCache cache;
private readonly CultNetDocumentRegistry registry;
private readonly string schemaId;
private readonly uint connectionId;
private readonly int maxFragmentBytes;
private readonly int maxPendingReliablePackets;
private readonly Dictionary<string, RelayPeer> peers = new(StringComparer.Ordinal);
public RudpMediaRelay(
IPEndPoint bind,
CultCache cache,
CultNetDocumentRegistry registry,
string schemaId,
uint connectionId,
int maxFragmentBytes,
int maxPendingReliablePackets)
{
this.cache = cache;
this.registry = registry;
this.schemaId = schemaId;
this.connectionId = connectionId;
this.maxFragmentBytes = maxFragmentBytes;
this.maxPendingReliablePackets = maxPendingReliablePackets;
socket = new Socket(bind.AddressFamily, SocketType.Dgram, ProtocolType.Udp);
socket.Bind(bind);
socket.Blocking = false;
}
public async Task RunAsync()
{
while (true)
{
ReceiveAvailable();
PumpReliableResends();
await Task.Delay(2).ConfigureAwait(false);
}
}
public void Dispose()
{
socket.Dispose();
}
private void ReceiveAvailable()
{
while (socket.Poll(0, SelectMode.SelectRead))
{
var buffer = new byte[65535];
EndPoint remote = new IPEndPoint(IPAddress.Any, 0);
int received;
try
{
received = socket.ReceiveFrom(buffer, ref remote);
}
catch (SocketException error) when (error.SocketErrorCode == SocketError.WouldBlock)
{
return;
}
var wire = buffer.AsSpan(0, received).ToArray();
var packet = CultNetRudpPacketCodec.Decode(wire);
if (packet.ConnectionId != connectionId)
{
continue;
}
var peer = GetPeer((IPEndPoint)remote);
if (packet.PacketType == CultNetRudpPacketType.Connect)
{
SendPacket(peer, peer.Session.AcceptConnect(packet, NowMs()));
continue;
}
var result = peer.Session.Receive(packet, NowMs());
if (result.Reply != null)
{
SendPacket(peer, result.Reply);
}
if (result.Disconnected)
{
peers.Remove(peer.Key);
continue;
}
foreach (var frame in result.Delivered.Where(static frame => string.Equals(frame.ChannelId, "schema", StringComparison.Ordinal)))
{
HandleSchemaFrameAsync(peer, frame.Payload).GetAwaiter().GetResult();
}
if (packet.PacketType == CultNetRudpPacketType.Accept || result.Delivered.Count > 0)
{
SendPacket(peer, peer.Session.CreateAck());
}
}
}
private async Task HandleSchemaFrameAsync(RelayPeer peer, byte[] payload)
{
var message = CultNetSchemaMessageSerialization.Deserialize(payload);
switch (message)
{
case CultNetDocumentPutRawMessage put:
await ApplyPutAsync(put).ConfigureAwait(false);
break;
case CultNetDatabaseSubscribeMessage subscribe:
peer.Subscriptions[subscribe.SubscriptionId] = subscribe;
Console.WriteLine($"relay RUDP subscribe peer={peer.Key} subscription={subscribe.SubscriptionId} schemas={string.Join(",", subscribe.SchemaIds ?? [])}");
break;
case CultNetDatabaseUnsubscribeMessage unsubscribe:
peer.Subscriptions.Remove(unsubscribe.SubscriptionId);
break;
}
}
private async Task ApplyPutAsync(CultNetDocumentPutRawMessage put)
{
if (put.Document == null ||
!string.Equals(put.Document.SchemaId, schemaId, StringComparison.Ordinal))
{
return;
}
await registry.ApplyRawDocumentPutMessageAsync(cache, put).ConfigureAwait(false);
await cache.FlushAsync(soft: true).ConfigureAwait(false);
var change = new CultNetDatabaseChangeRawMessage
{
MessageId = Guid.NewGuid().ToString("N"),
ChangeKind = "updated",
Document = put.Document
};
var frame = CultNetSchemaMessageSerialization.Serialize(change);
foreach (var peer in peers.Values)
{
foreach (var subscription in peer.Subscriptions.Values)
{
if (!Matches(subscription, put.Document.SchemaId, put.Document.RecordKey))
{
continue;
}
change.SubscriptionId = subscription.SubscriptionId;
SendSchema(peer, CultNetSchemaMessageSerialization.Serialize(change));
if (!peer.HasForwardedFrame)
{
peer.HasForwardedFrame = true;
Console.WriteLine($"relay RUDP forward peer={peer.Key} subscription={subscription.SubscriptionId} schema={put.Document.SchemaId}");
}
ReceiveAvailable();
}
}
var document = (MimirCultMeshMediaFrameDocument)registry
.GetBySchemaId(schemaId)!
.PayloadDeserializer(put.Document.Payload);
if (document.Sequence % 120 == 0)
{
Console.WriteLine($"relay RUDP put stream={document.StreamId} sequence={document.Sequence} bytes={document.PayloadBytes}");
}
}
private static bool Matches(CultNetDatabaseSubscribeMessage request, string schemaId, string recordKey)
{
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(recordKey, StringComparer.Ordinal);
return schemaMatches && keyMatches;
}
private RelayPeer GetPeer(IPEndPoint endpoint)
{
var key = endpoint.ToString();
if (peers.TryGetValue(key, out var existing))
{
return existing;
}
var peer = new RelayPeer(
key,
endpoint,
new CultNetRudpSession(new CultNetRudpSessionOptions
{
ConnectionId = connectionId,
InitialSequence = (uint)(peers.Count + 1),
ResendDelayMs = 100,
MaxPendingReliablePackets = maxPendingReliablePackets
}));
peers[key] = peer;
return peer;
}
private void SendSchema(RelayPeer peer, byte[] payload)
{
foreach (var packet in peer.Session.SendMany(
"schema",
payload,
new CultNetRudpSendOptions { Reliable = true, Ordered = true, NowMs = NowMs() },
maxFragmentBytes))
{
SendPacket(peer, packet);
}
}
private void SendPacket(RelayPeer peer, CultNetRudpPacket packet)
{
var wire = CultNetRudpPacketCodec.Encode(packet);
socket.SendTo(wire, peer.Endpoint);
}
private void PumpReliableResends()
{
foreach (var peer in peers.Values)
{
foreach (var packet in peer.Session.DueResends(NowMs()))
{
SendPacket(peer, packet);
}
}
}
private static long NowMs() => DateTimeOffset.UtcNow.ToUnixTimeMilliseconds();
}
sealed record RelayPeer(
string Key,
IPEndPoint Endpoint,
CultNetRudpSession Session)
{
public Dictionary<string, CultNetDatabaseSubscribeMessage> Subscriptions { get; } = new(StringComparer.Ordinal);
public bool HasForwardedFrame { get; set; }
}
sealed class CliOptions
{
private readonly Dictionary<string, string> values = new(StringComparer.OrdinalIgnoreCase);
public static CliOptions Parse(IEnumerable<string> args)
{
var parsed = new CliOptions();
var pendingKey = string.Empty;
foreach (var arg in args)
{
if (arg.StartsWith("--", StringComparison.Ordinal))
{
pendingKey = arg[2..];
parsed.values[pendingKey] = "true";
continue;
}
if (!string.IsNullOrWhiteSpace(pendingKey))
{
parsed.values[pendingKey] = arg;
pendingKey = string.Empty;
}
}
return parsed;
}
public string? Get(string key) => values.TryGetValue(key, out var value) ? value : null;
public int GetInt(string key, int fallback) =>
int.TryParse(Get(key), NumberStyles.Integer, CultureInfo.InvariantCulture, out var value) ? value : fallback;
public uint GetUInt(string key, uint fallback) =>
uint.TryParse(Get(key), NumberStyles.Integer, CultureInfo.InvariantCulture, out var value) ? value : fallback;
}

View file

@ -0,0 +1,24 @@
using GameCult.Caching;
using MessagePack;
namespace Mimir.Runtime.Synchronization;
[CultDocument("mimir.cultmesh_media_frame", "mimir.cultmesh_media_frame.v1")]
[MessagePackObject]
public sealed record MimirCultMeshMediaFrameDocument(
[property: Key(0)]
[property: CultName]
string FrameId,
[property: Key(1)] string StreamId,
[property: Key(2)] long Sequence,
[property: Key(3)] string ProducedAtUtc,
[property: Key(4)] long TimestampNanoseconds,
[property: Key(5)] string PayloadKind,
[property: Key(6)] string Container,
[property: Key(7)] string VideoCodec,
[property: Key(8)] string AudioCodec,
[property: Key(9)] int PayloadBytes,
[property: Key(10)] byte[] Payload,
[property: Key(11)] string ProducerNode,
[property: Key(12)] string ClockDomainId,
[property: Key(13)] string[] Tags);

View file

@ -6,12 +6,14 @@ public enum MimirNetworkPayloadKind
RawAudioDebugWindow,
CompressedAudioProgramFeed,
CompressedVideoProgramFeed,
RawVideoDiagnosticWindow
RawVideoDiagnosticWindow,
MuxedAudioVideoProgramFeed
}
public enum MimirNetworkTransportKind
{
CultMeshTypedState,
CultMeshReliableUdpMedia,
LanUdpDatagrams,
SrtDiagnosticMedia,
WebRtcExperimental,
@ -68,6 +70,18 @@ public static class MimirNetworkTransportConfigurations
RejectionConditions: ["used-as-clock-authority", "used-for-local-hot-path"],
"Useful bridge. Not the Perfect Machine's synchronization core.");
public static MimirNetworkTransportConfiguration CultMeshProgramBridge { get; } = new(
"cultmesh-program-bridge",
"Reliable-UDP CultMesh media bridge for muxed program frames that OBS consumes through a local Starfire endpoint.",
MimirNetworkTransportKind.CultMeshReliableUdpMedia,
MimirNetworkPayloadKind.MuxedAudioVideoProgramFeed,
MayAffectClock: false,
CarriesRawMedia: true,
TargetLatencyMilliseconds: 500.0,
RequiredDocuments: ["mimir.cultmesh_media_frame"],
RejectionConditions: ["used-as-clock-authority", "unbounded-media-cache", "non-local-obs-egress"],
"Raven publishes rolling media-frame documents over CultNet reliable UDP; Starfire lowers them to a local OBS MPEG-TS endpoint.");
public static MimirNetworkTransportConfiguration WebRtcExperiment { get; } = CultMeshTimingState with
{
Id = "webrtc-experimental-media",
@ -84,6 +98,7 @@ public static class MimirNetworkTransportConfigurations
[
CultMeshTimingState,
RawAudioDebugWindow,
CultMeshProgramBridge,
SrtProgramBridge,
WebRtcExperiment
];

View file

@ -142,3 +142,4 @@
{"ts":"2026-06-16T00:00:00+01:00","type":"correction","status":"ok","note":"Reoriented Mimir's daemon-swarm map after Odin/CultLib RUDP work: Raven media body transport remains src/Mimir.CultMeshMedia over CultNet/CultMesh reliable UDP with OBS-local UDP as egress only, while Eve dashboard/browser reference health now publishes idunn.daemon_health over cultnet.transport.rudp.v0 as self-witness state. Odin owns accepted Verse/service discovery and Idunn owns keepalive decisions; Mimir health publishers do not own lifecycle."}
{"ts":"2026-06-16T00:05:00+01:00","type":"correction","status":"ok","note":"Corrected the transport model: CultLib has added authorized RUDP helpers across runtimes and Odin/Idunn now treats daemon truth as typed CultCache/CultMesh state carried over cultnet.transport.rudp.v0. Mimir should not assume TCP is needed for any daemon-truth path once the runtime has RUDP support; HTTP/WebSocket/TCP are client lowerings, compatibility probes, or migration debt. Health is separate from media by authority, not because it needs a separate TCP transport."}
{"ts":"2026-06-16T00:12:00+01:00","type":"correction","status":"ok","note":"Source audit found Mimir.CultMeshMedia still using older C# CultMesh.StartNodeAsync/CultMesh.ConnectClient entrypoints while CultLib now exposes C# CultMesh.CreateRudpServer/CreateRudpClient/CreateRudpClientForAuthorizedPeer. Updated maps to treat the media bridge's current transport path as migration debt: typed program/media documents should move to explicit CultNet RUDP helpers, with OBS-local UDP preserved only as compatibility egress."}
{"ts": "2026-06-16T18:05:56+01:00", "type": "verification", "status": "ok", "note": "Mimir.CultMeshMedia migrated from old CultMesh node/client transport to explicit CultNet RUDP relay/send/recv path; local smoke artifacts/rudp-media-smoke-20260616-180404 proved 32 MPEG-TS chunks reached Starfire-local UDP sink over RUDP (packets=32 bytes=42112)."}

View file

@ -383,12 +383,10 @@ accepted_design:
`mimir.cultmesh_media_frame` documents over CultNet reliable UDP,
Yggdrasil relays them on `10.77.0.1:3075`, and Starfire lowers the
subscribed stream to `udp://127.0.0.1:5200` for the OBS source
`Raven Monitor + Realtek`. Source audit warning: the current C# bridge
still calls older `CultMesh.StartNodeAsync`/`CultMesh.ConnectClient`
entrypoints; now that CultLib exposes C# RUDP facade helpers, the next
transport cut should move this bridge onto explicit RUDP server/client or
authorized-peer helpers and demote any TCP-like node/client path to
compatibility evidence.
`Raven Monitor + Realtek`. The bridge now uses explicit CultLib RUDP
socket/session helpers for relay, sender, and receiver. The old
`CultMesh.StartNodeAsync`/`CultMesh.ConnectClient` node/client path is no
longer a transport authority for this media lane.
- `src/Mimir.EveDashboard` publishes `mimir.eve_dashboard_state` through
CultNet/CultMesh and can push `idunn.daemon_health` records over
`cultnet.transport.rudp.v0` for itself and a paired service daemon. Its
@ -433,9 +431,11 @@ active_subgoals:
keepalive, and Mimir surfaces publish typed RUDP self-health rather than
private watchdog loops or TCP/HTTP status truth.
- Cut remaining older CultMesh node/client transport assumptions in
`src/Mimir.CultMeshMedia`, `src/Mimir.EveDashboard`, and witness tooling:
daemon truth and typed program/media document paths should use explicit
CultNet RUDP helpers; HTTP/WebSocket remain render/client lowerings only.
`src/Mimir.EveDashboard` and witness tooling: daemon truth, provider
catalogs, command boundaries, and typed program document paths should use
explicit CultNet RUDP helpers; HTTP/WebSocket remain render/client
lowerings only. `src/Mimir.CultMeshMedia` has completed this cut for the
Raven/Yggdrasil/Starfire media-frame lane.
current_status:
summary: >