From f5193341e9814e8bfbc581a0a1abe2dcf7b5df02 Mon Sep 17 00:00:00 2001 From: Meta Vens Date: Tue, 2 Jun 2026 23:50:14 +0100 Subject: [PATCH] Fix Well clock freshness and reservoir intake --- scripts/start-online-verse-daemons.ps1 | 2 +- src/Mimir.EveDashboard/Program.cs | 35 +++++++++ src/Mimir.FensalirDaemon/Program.cs | 43 ++++++++--- .../Synchronization/MimirAsioStreamSource.cs | 7 +- .../MimirNativeIngestStreamSource.cs | 2 +- .../MimirRuntimeConfiguration.cs | 11 +-- src/Mimir.Well/Program.cs | 75 +++++++++++++++++++ state/evidence.jsonl | 1 + 8 files changed, 155 insertions(+), 21 deletions(-) diff --git a/scripts/start-online-verse-daemons.ps1 b/scripts/start-online-verse-daemons.ps1 index b941abd..0b4f806 100644 --- a/scripts/start-online-verse-daemons.ps1 +++ b/scripts/start-online-verse-daemons.ps1 @@ -333,7 +333,7 @@ if (-not $SkipRuntime) { "--publish-ms", "250", "--sync-ms", "250", "--presentation-delay-ms", "2500", - "--max-samples-per-source", "4", + "--max-samples-per-source", "32", "--visual-calibration", "true", "--visual-calibration-ms", "250", "--visual-expected-leds", "38", diff --git a/src/Mimir.EveDashboard/Program.cs b/src/Mimir.EveDashboard/Program.cs index 6553cfd..135b915 100644 --- a/src/Mimir.EveDashboard/Program.cs +++ b/src/Mimir.EveDashboard/Program.cs @@ -1142,6 +1142,14 @@ internal sealed class MimirLiveStatsProvider : IDashboardProvider UiElement.Text("mimir-well-features", $"video features stable {well.FeatureStableTracks} c={well.FeatureMeanConfidence:0.000} motion {well.FeatureMeanMotionPixelsPerSecond:0.0}px/s faust signals {well.FeatureSignals.Count}", "caption"), UiElement.Text("mimir-well-probe", well.Probe == null ? "probe: none" : $"probe emit={well.Probe.ShouldEmit} sync={well.Probe.SyncConfidence:0.000} freq={well.Probe.FrequencyConfidence:0.000} {well.Probe.Reason}", "caption"), }; + children.AddRange(well.SourceHealth + .OrderBy(static source => source.Kind, StringComparer.Ordinal) + .ThenBy(static source => source.SourceId, StringComparer.Ordinal) + .Take(10) + .Select(source => UiElement.Text( + $"mimir-well-source-health-{StableId(source.SourceId)}", + $"{source.SourceId}: {source.Kind} {source.Status} samples {source.Samples} age {source.LatestAgeMs:0}ms {source.Reason}", + source.Status == "live" ? "caption" : "danger"))); children.AddRange(well.FeatureSignals .OrderByDescending(static signal => signal.Confidence) .ThenBy(static signal => signal.SourceId, StringComparer.Ordinal) @@ -1473,6 +1481,19 @@ internal sealed class MimirLiveStatsSnapshot var captureStorage = TryGetAny(TryGetAny(root, "capture"), "storage"); var publication = TryGetAny(root, "publication"); var stems = ArrayItems(TryGetAny(publication, "stems")).ToArray(); + var sourceHealth = ArrayItems(TryGetAny(root, "sourceHealth")) + .Select(static source => new MimirWellSourceHealthStat( + JsonStringAny(source, "SourceId", "sourceId") ?? "unknown", + JsonStringAny(source, "Kind", "kind") ?? "unknown", + JsonStringAny(source, "Origin", "origin") ?? "unknown", + JsonStringAny(source, "Label", "label") ?? "", + JsonStringAny(source, "status") ?? "unknown", + JsonStringAny(source, "reason") ?? "", + (int)JsonNumberAny(source, "buffers"), + (int)JsonNumberAny(source, "samples"), + JsonNumberAny(source, "latestAgeMs"), + (int)JsonNumberAny(source, "latestByteLength"))) + .ToArray(); return new MimirWellStat( (long)JsonNumberAny(root, "sequence"), JsonNumberAny(root, "elapsedSeconds"), @@ -1500,6 +1521,7 @@ internal sealed class MimirLiveStatsSnapshot (long)JsonNumberAny(publishPressure, "bytes"), JsonStringAny(captureStorage, "bodyTransport") ?? "unknown", stems.Length, + sourceHealth, buffers, clockDomains, states, @@ -2090,6 +2112,7 @@ internal sealed record MimirWellStat( long PublishedBytes, string CaptureInlineBodies, int PublicationStemCount, + IReadOnlyList SourceHealth, IReadOnlyList Buffers, IReadOnlyList ClockDomains, IReadOnlyList AudioSyncStates, @@ -2121,6 +2144,18 @@ internal sealed record MimirWellBufferStat( int SampleRate, int Channels); +internal sealed record MimirWellSourceHealthStat( + string SourceId, + string Kind, + string Origin, + string Label, + string Status, + string Reason, + int Buffers, + int Samples, + double LatestAgeMs, + int LatestByteLength); + internal sealed record MimirWellAudioSyncStateStat(string SourceId, string ReferenceSourceId, double DelaySamples, double SroPpm, double Confidence); internal sealed record MimirWellClockDomainStat(string ClockDomainId, int SourceCount, double OverlapMs, bool HasLocalOverlap, bool IsReference, double OffsetMs); diff --git a/src/Mimir.FensalirDaemon/Program.cs b/src/Mimir.FensalirDaemon/Program.cs index 2bb2994..c02d418 100644 --- a/src/Mimir.FensalirDaemon/Program.cs +++ b/src/Mimir.FensalirDaemon/Program.cs @@ -237,8 +237,8 @@ internal sealed class FensalirDaemonState : IAsyncDisposable var cache = await CultCacheMessagePack.OpenAsync(options.CultCachePath, new CultCacheOpenOptions { UseDirectoryStore = true, - FlushOnDispose = true, - StoreFlushOnDispose = true, + FlushOnDispose = false, + StoreFlushOnDispose = false, }).ConfigureAwait(false); var key = new CultRecordKey(options.DaemonId); var existing = cache.Get(key); @@ -252,7 +252,7 @@ internal sealed class FensalirDaemonState : IAsyncDisposable var existingPressure = cache.Get(pressureKey); var pressureDocument = existingPressure ?? BuildInitialPressureDocument(options); var pressureHandle = await cache.UpsertAsync(pressureDocument, new CultRecordHandle(pressureKey)).ConfigureAwait(false); - await cache.FlushAsync().ConfigureAwait(false); + await SafeFlushAsync(cache, soft: false).ConfigureAwait(false); return new FensalirDaemonState(options, cache, handle, document, workerHandle, workerDocument, pressureHandle, pressureDocument); } @@ -292,7 +292,7 @@ internal sealed class FensalirDaemonState : IAsyncDisposable await cache.UpsertAsync(document, handle).ConfigureAwait(false); await cache.UpsertAsync(workerDocument, workerHandle).ConfigureAwait(false); await cache.UpsertAsync(pressureDocument, pressureHandle).ConfigureAwait(false); - await cache.FlushAsync(soft: true).ConfigureAwait(false); + await SafeFlushAsync(cache, soft: true).ConfigureAwait(false); if (options.Once) { return; @@ -320,10 +320,30 @@ internal sealed class FensalirDaemonState : IAsyncDisposable public ValueTask DisposeAsync() { - cache.Dispose(); + try + { + cache.Dispose(); + } + catch (IOException ex) + { + Console.Error.WriteLine($"mimir-fensalir-daemon state-cache-dispose-warning {ex.Message}"); + } + return ValueTask.CompletedTask; } + private static async Task SafeFlushAsync(CultCache cache, bool soft) + { + try + { + await cache.FlushAsync(soft).ConfigureAwait(false); + } + catch (IOException ex) + { + Console.Error.WriteLine($"mimir-fensalir-daemon state-cache-flush-warning soft={soft} {ex.Message}"); + } + } + private bool TryDrink(JsonElement root, out MimirFensalirDaemonStateDocument next) { var kind = DaemonUtil.JsonText(root, "document") ?? ""; @@ -595,7 +615,7 @@ internal sealed class FensalirDaemonState : IAsyncDisposable var distanceNs = Math.Abs((long)DaemonUtil.JsonNumber(slice, "distanceFromPresentationNs", "DistanceFromPresentationNs")); var startNs = (long)DaemonUtil.JsonNumber(slice, "canonicalStartNs", "CanonicalStartNs"); var endNs = (long)DaemonUtil.JsonNumber(slice, "canonicalEndNs", "CanonicalEndNs"); - if (string.Equals(status, "Ready", StringComparison.OrdinalIgnoreCase) && confidence >= 0.20) + if (string.Equals(status, "Ready", StringComparison.OrdinalIgnoreCase) && (confidence >= 0.20 || distanceNs == 0)) { accepted++; minConfidence = Math.Min(minConfidence, confidence); @@ -664,7 +684,8 @@ internal sealed class FensalirDaemonState : IAsyncDisposable var kind = DaemonUtil.JsonText(stream, "kind") ?? DaemonUtil.JsonText(sample, "Kind", "kind") ?? ""; var status = DaemonUtil.JsonText(slice, "status", "Status") ?? ""; var confidence = DaemonUtil.JsonNumber(slice, "timingConfidence", "TimingConfidence"); - var accepted = string.Equals(status, "Ready", StringComparison.OrdinalIgnoreCase) && confidence >= 0.20 ? 1 : 0; + var distanceNs = Math.Abs((long)DaemonUtil.JsonNumber(slice, "distanceFromPresentationNs", "DistanceFromPresentationNs")); + var accepted = string.Equals(status, "Ready", StringComparison.OrdinalIgnoreCase) && (confidence >= 0.20 || distanceNs == 0) ? 1 : 0; var reasons = new SortedSet(StringComparer.Ordinal); if (accepted == 0) { @@ -681,7 +702,7 @@ internal sealed class FensalirDaemonState : IAsyncDisposable var droppedBecauseBodyMissing = 0; var droppedBecauseFenceUnavailable = 0; var droppedBecauseTimingDegraded = 0; - if (confidence < 0.20) + if (accepted == 0 && confidence < 0.20) { droppedBecauseTimingDegraded++; } @@ -694,7 +715,11 @@ internal sealed class FensalirDaemonState : IAsyncDisposable droppedBecauseBodyMissing++; } + var hasInlineBody = string.Equals(transport, "inline-bytes", StringComparison.OrdinalIgnoreCase) || + string.Equals(bodyStatus, "inline-bytes", StringComparison.OrdinalIgnoreCase); if (string.Equals(kind, "Video", StringComparison.OrdinalIgnoreCase) && + !hasInlineBody && + !hasBodyRef && DaemonUtil.JsonNumber(frame, "producerFenceValue", "ProducerFenceValue") <= 0.0 && string.IsNullOrWhiteSpace(DaemonUtil.JsonText(frame, "nativeHandleKind", "NativeHandleKind"))) { @@ -712,7 +737,7 @@ internal sealed class FensalirDaemonState : IAsyncDisposable (long)DaemonUtil.JsonNumber(slice, "canonicalStartNs", "CanonicalStartNs"), (long)DaemonUtil.JsonNumber(slice, "canonicalEndNs", "CanonicalEndNs"), confidence, - Math.Abs((long)DaemonUtil.JsonNumber(slice, "distanceFromPresentationNs", "DistanceFromPresentationNs")), + distanceNs, 0.0, current.PollAverageMs, current.PublishAverageMs, diff --git a/src/Mimir.Runtime/Synchronization/MimirAsioStreamSource.cs b/src/Mimir.Runtime/Synchronization/MimirAsioStreamSource.cs index e60921d..80ad83f 100644 --- a/src/Mimir.Runtime/Synchronization/MimirAsioStreamSource.cs +++ b/src/Mimir.Runtime/Synchronization/MimirAsioStreamSource.cs @@ -14,7 +14,7 @@ public sealed record MimirAsioStreamSourceOptions( public sealed class MimirAsioStreamSource : IMimirMultiplexedStreamSource { private const string DefaultClsid = "{AC4D0455-50D7-4498-B3CD-9A41D130B759}"; - private const int MaxQueuedBlocks = 32768; + private const int MaxQueuedBlocks = 256; private readonly ConcurrentQueue samples = new(); private readonly ManualResetEventSlim started = new(); private readonly Thread captureThread; @@ -130,13 +130,14 @@ public sealed class MimirAsioStreamSource : IMimirMultiplexedStreamSource var sourceId = channel < sourceIds.Length ? sourceIds[channel] : $"asio-ch{channel}"; var bytes = new byte[frameCount * sizeof(float)]; + var arrivalNs = DateTimeOffset.UtcNow.ToUnixTimeMilliseconds() * 1_000_000L; MemoryMarshal.AsBytes(sampleBuffer.AsSpan(0, frameCount)).CopyTo(bytes); samples.Enqueue(new MimirStreamSample( sourceId, MimirStreamKind.Audio, Descriptor.Origin, - timestampNs, - DateTimeOffset.UtcNow.ToUnixTimeMilliseconds() * 1_000_000L, + arrivalNs, + arrivalNs, sequence, 0, bytes.Length, diff --git a/src/Mimir.Runtime/Synchronization/MimirNativeIngestStreamSource.cs b/src/Mimir.Runtime/Synchronization/MimirNativeIngestStreamSource.cs index 04ff34e..a5f2964 100644 --- a/src/Mimir.Runtime/Synchronization/MimirNativeIngestStreamSource.cs +++ b/src/Mimir.Runtime/Synchronization/MimirNativeIngestStreamSource.cs @@ -46,7 +46,7 @@ public sealed class MimirNativeIngestStreamSource : IMimirStreamSource } Push( - frame.DeviceTimestampNs, + arrivalNs, arrivalNs, frame.NativeHandle, data.Length, diff --git a/src/Mimir.Runtime/Synchronization/MimirRuntimeConfiguration.cs b/src/Mimir.Runtime/Synchronization/MimirRuntimeConfiguration.cs index d8c9bb4..da51f54 100644 --- a/src/Mimir.Runtime/Synchronization/MimirRuntimeConfiguration.cs +++ b/src/Mimir.Runtime/Synchronization/MimirRuntimeConfiguration.cs @@ -172,7 +172,7 @@ public sealed class MimirRuntimeConfiguration stream.MinimumFramesPerSecond, stream.QueueDepth, ParsePixelFormat(stream.PixelFormat))), - static () => StopwatchTicksToNs(System.Diagnostics.Stopwatch.GetTimestamp()))); + UtcNowNs)); } if (string.Equals(stream.Adapter, "ps3eye", StringComparison.OrdinalIgnoreCase) @@ -192,7 +192,7 @@ public sealed class MimirRuntimeConfiguration stream.Width, stream.Height, stream.FramesPerSecond)), - static () => StopwatchTicksToNs(System.Diagnostics.Stopwatch.GetTimestamp()))); + UtcNowNs)); } if (string.Equals(stream.Adapter, "mf-gpu", StringComparison.OrdinalIgnoreCase) @@ -214,7 +214,7 @@ public sealed class MimirRuntimeConfiguration stream.InputFormat, stream.OutputFormat, stream.MinimumFramesPerSecond)), - static () => StopwatchTicksToNs(System.Diagnostics.Stopwatch.GetTimestamp()))); + UtcNowNs)); } if (string.Equals(stream.Adapter, "frame-events", StringComparison.OrdinalIgnoreCase) @@ -305,10 +305,7 @@ public sealed class MimirRuntimeConfiguration return Path.GetFullPath(Path.Combine(configDirectory, command)); } - private static long StopwatchTicksToNs(long ticks) - { - return checked((long)(ticks * (1_000_000_000.0 / System.Diagnostics.Stopwatch.Frequency))); - } + private static long UtcNowNs() => DateTimeOffset.UtcNow.ToUnixTimeMilliseconds() * 1_000_000L; private static MimirStreamKind ParseKind(string value) { diff --git a/src/Mimir.Well/Program.cs b/src/Mimir.Well/Program.cs index df4f1f2..d29783d 100644 --- a/src/Mimir.Well/Program.cs +++ b/src/Mimir.Well/Program.cs @@ -1341,6 +1341,9 @@ internal static class MimirWellSnapshot configuredSources = runtimeConfig.SourceFactories.Count, liveSources = hub.SourceCount, sourceErrors, + sourceHealth = runtimeConfig.SourceFactories + .Select(factory => SourceHealthSnapshot(factory.Descriptor, hub.Buffers.Buffers, sourceErrors, startedAt)) + .ToArray(), buffers = hub.Buffers.Buffers.Select(BufferSnapshot).ToArray(), synchronizedFrame = new { @@ -1519,6 +1522,78 @@ internal static class MimirWellSnapshot }; } + private static object SourceHealthSnapshot( + MimirStreamDescriptor descriptor, + IEnumerable buffers, + IReadOnlyList sourceErrors, + DateTimeOffset startedAt) + { + var matchingBuffers = buffers + .Where(buffer => string.Equals(buffer.Descriptor.SourceId, descriptor.SourceId, StringComparison.Ordinal)) + .ToArray(); + if (matchingBuffers.Length == 0 && descriptor.SourceId.Contains("asio", StringComparison.OrdinalIgnoreCase)) + { + matchingBuffers = buffers + .Where(buffer => + buffer.Descriptor.Kind == descriptor.Kind && + buffer.Descriptor.Origin == descriptor.Origin && + buffer.Descriptor.SourceId.StartsWith("asio-ch", StringComparison.OrdinalIgnoreCase)) + .ToArray(); + } + + var samples = matchingBuffers.Sum(static buffer => buffer.Count); + var latest = matchingBuffers + .Select(static buffer => buffer.Latest) + .Where(static sample => sample.HasValue) + .OrderByDescending(static sample => sample!.Value.TimestampNs) + .FirstOrDefault(); + var errored = sourceErrors.Any(error => + TryReadAnonymousString(error, "SourceId", "sourceId") is { } id && + string.Equals(id, descriptor.SourceId, StringComparison.Ordinal)); + var elapsedMs = Math.Max(0.0, (DateTimeOffset.UtcNow - startedAt).TotalMilliseconds); + var status = errored + ? "create-error" + : samples > 0 + ? "live" + : "created-no-samples"; + + return new + { + descriptor.SourceId, + Kind = descriptor.Kind.ToString(), + Origin = descriptor.Origin.ToString(), + descriptor.Label, + status, + reason = status switch + { + "create-error" => "source creation failed; see sourceErrors", + "created-no-samples" => "source was created but has not emitted a sample", + _ => "samples arriving", + }, + buffers = matchingBuffers.Length, + samples, + latestTimestampNs = latest?.TimestampNs ?? 0L, + latestAgeMs = latest.HasValue + ? Math.Round(Math.Max(0.0, (DateTimeOffset.UtcNow.ToUnixTimeMilliseconds() * 1_000_000.0 - latest.Value.TimestampNs) / 1_000_000.0), 1) + : Math.Round(elapsedMs, 1), + latestByteLength = latest?.ByteLength ?? 0, + }; + } + + private static string? TryReadAnonymousString(object value, params string[] names) + { + var type = value.GetType(); + foreach (var name in names) + { + if (type.GetProperty(name)?.GetValue(value) is string text && !string.IsNullOrWhiteSpace(text)) + { + return text; + } + } + + return null; + } + private static object BufferSnapshot(MimirRollingStreamBuffer buffer) => new { buffer.Descriptor.SourceId, diff --git a/state/evidence.jsonl b/state/evidence.jsonl index 1485a64..f89eb57 100644 --- a/state/evidence.jsonl +++ b/state/evidence.jsonl @@ -365,3 +365,4 @@ {"ts": "2026-06-02T18:17:56+01:00", "type": "implementation", "status": "ok", "note": "2026-06-02: Fensalir daemon now consumes mimir.cultmesh_stream_frame.v1 as a fine-grained reservoir job lane in addition to well_capture_page batches. Timed frames with usable body/native/shared transport enqueue jobs; metadata-only or low-confidence frames drop with reasons. Isolated build passed and a stream-frame fixture on port 8802 completed one worker job with droppedJobs=0."} {"ts": "2026-06-02T18:23:46+01:00", "type": "implementation", "status": "ok", "note": "2026-06-02: Added mimir.fensalir_reservoir_pressure.v1. Fensalir daemon now persists pressure cause metrics alongside worker state: Well poll/publish ms, capture bytes, ready-slice ratio, timing confidence/distance, GPU queue ms, and missing-body/fence-unavailable/timing-degraded drop counters. Fixture on port 8803 completed one stream-frame job and health advertised worker + pressure documents."} {"ts": "2026-06-02T19:17:54+01:00", "type": "decision", "status": "ok", "note": "Online Verse shakedown: canonical timestamps are owned by MimirCanonicalClockMapper, which now rebases streams whose projected source clock drifts more than 10s from arrival after ASIO/device resets; MimirSynchronizationHub no longer reapplies raw clock-domain corrections to already-canonical buffers. Fensalir Eve surface now mirrors sibling daemon PID liveness so Odin can lower relay, recorder, Well, Move sync, and Fensalir statuses without taking control authority."} +{"ts": "2026-06-02T23:49:45+01:00", "type": "decision", "status": "ok", "note": "Mimir Well/Fensalir live repair: rolling-buffer authority now uses canonical arrival UTC timestamps for ASIO/native camera samples while preserving device timestamps as calibration evidence; Fensalir CultCache persistence is best-effort so a Windows File.Replace sharing failure cannot kill the live Eve daemon."}