using System.Text.Json; using Fleet.Core; using MoShiro.Protocol; namespace MoShiro.Measurement; /// Agent 单调时钟(微秒)与 UTC(毫秒)的对应关系:块头里的时钟锚点。 public readonly record struct ClockAnchor(long ListenerMonoUs, long UtcMs); /// /// 给合并好的流盖章:生成完整的块头(任务/测试、参数快照、端口、时钟锚点、起止时间)。 /// 起止时间由数据本身的时间戳换算得出,而不是取收到列块的时刻,所以不受拨号方交付延迟的影响。 /// public static class ChunkFinalizer { public const int MaxDurationMs = 600_000; public static RawChunk Finalize( KindSpec spec, RawChunk fragment, IReadOnlyList streams, NodeId agent, ClockAnchor anchor, IReadOnlyDictionary resolvedParams, int port, ChunkId id) { RawChunkHeader source = fragment.Header; long offsetUs = source.Clock.DialerOffsetUs; long min = long.MaxValue; long max = long.MinValue; foreach (RawStream stream in streams) { if (stream.Name == "tcp_info") { continue; // 这个流的时钟取决于 side 列,不参与起止时间的计算 } foreach (RawColumn column in stream.Columns) { if (column.Role == ColumnRole.None) { continue; } for (int i = 0; i < stream.Count; i++) { if (!column.IsValid(i)) { continue; } long utc = ToUtcMs(column.Values[i], column.Role, anchor, offsetUs); min = Math.Min(min, utc); max = Math.Max(max, utc); } } } long start = min == long.MaxValue ? anchor.UtcMs - source.DurationMs : min; int duration = (int)Math.Clamp(Math.Max(source.DurationMs, min == long.MaxValue ? 0 : max - start), 1, MaxDurationMs); var header = new RawChunkHeader { ChunkId = id, Kind = spec.Id, TaskId = source.TaskId, TestId = source.TestId, Phase = source.Phase, AgentId = agent, DialerId = source.DialerId, IpVersion = source.IpVersion, Target = null, Params = new Dictionary(resolvedParams), Port = port, Direction = source.Direction, StartUtcMs = start, DurationMs = duration, Partial = false, Clock = new ChunkClock { ListenerMonoUs = anchor.ListenerMonoUs, UtcMs = anchor.UtcMs, DialerOffsetUs = offsetUs }, }; return new RawChunk(header, streams); } public static long ToUtcMs(long value, ColumnRole role, ClockAnchor anchor, long dialerOffsetUs) { long listenerUs = role == ColumnRole.TsDialer ? value + dialerOffsetUs : value; long delta = listenerUs - anchor.ListenerMonoUs; return anchor.UtcMs + (delta >= 0 ? delta / 1000 : -((-delta + 999) / 1000)); } }