diff --git a/backend/src/MoShiro.Backend.Analysis/Engine/TestEvaluator.cs b/backend/src/MoShiro.Backend.Analysis/Engine/TestEvaluator.cs index 05c769d..1d78a4d 100644 --- a/backend/src/MoShiro.Backend.Analysis/Engine/TestEvaluator.cs +++ b/backend/src/MoShiro.Backend.Analysis/Engine/TestEvaluator.cs @@ -31,7 +31,7 @@ public sealed class TestResult /// /// 测试结果的计算：所有阶段的块到达之后，按方案算出指标和曲线。 -/// udp_train 的各个阶段作为码率阶梯：可持续码率、有效吞吐、限速开始的位置。 +/// quic_stream 的各个阶段作为码率阶梯：可持续码率、有效吞吐、限速开始的位置。 /// public static class TestEvaluator { @@ -43,7 +43,7 @@ public static class TestEvaluator { var result = new TestResult(); EvaluateMetrics(chunks, result); - if (chunks.Any(c => c.Header.Kind == Kinds.UdpTrain.Id)) + if (chunks.Any(c => c.Header.Kind == Kinds.QuicStream.Id)) { EvaluateLadder(chunks, result); } @@ -95,7 +95,7 @@ public static class TestEvaluator // 一个阶段（一档码率）可能被切成好几个块（一个块最多 50000 条记录，100 Mbit/s 的 10 秒要切 3～4 块）： // 阶梯里的一步是一个阶段，不是一个块——同一阶段的块先合并，再算丢包、抖动和实际吞吐。 - foreach (IGrouping phase in chunks.Where(c => c.Header.Kind == Kinds.UdpTrain.Id).GroupBy(c => c.Header.Phase ?? 0).OrderBy(g => g.Key)) + foreach (IGrouping phase in chunks.Where(c => c.Header.Kind == Kinds.QuicStream.Id).GroupBy(c => c.Header.Phase ?? 0).OrderBy(g => g.Key)) { RawChunk[] parts = [.. phase.Where(c => c.Find("pkt") is not null).OrderBy(c => c.Header.StartUtcMs)]; if (parts.Length == 0) diff --git a/backend/src/MoShiro.Backend.Analysis/Metrics/HttpMetrics.cs b/backend/src/MoShiro.Backend.Analysis/Metrics/HttpMetrics.cs new file mode 100644 index 0000000..eb9e448 --- /dev/null +++ b/backend/src/MoShiro.Backend.Analysis/Metrics/HttpMetrics.cs @@ -0,0 +1,48 @@ +namespace MoShiro.Backend.Analysis; + +/// http 任务（单个 Agent 请求一个 URL）的指标。耗时只统计成功的请求。 +public static class HttpMetrics +{ + private static readonly string[] s_req = ["http_request.req"]; + + public static readonly MetricDef Success = new( + "http.success", AccKind.Ratio, s_req, Ops.HttpSuccess, "HTTP success", "Share of requests that got an answer with an accepted status.", Better.Higher, + [new MetricExport("http.success", Stat.Pct, Unit.Pct, 99, 90)]); + + public static readonly MetricDef Dns = Metric.Histogram( + "http.dns_ms", s_req, Ops.HttpPhaseMs("dns_us"), "DNS time", "Time to resolve the target host name.", Unit.Ms, [Stat.P95], good: 50, poor: 300); + + public static readonly MetricDef Tcp = Metric.Histogram( + "http.tcp_ms", s_req, Ops.HttpPhaseMs("tcp_us"), "TCP connect time", "Time to establish the TCP connection (HTTP/1.1 and HTTP/2).", Unit.Ms, [Stat.P95], good: 100, poor: 400); + + public static readonly MetricDef Ttfb = Metric.Histogram( + "http.ttfb_ms", s_req, Ops.HttpPhaseMs("ttfb_us"), "Time to first byte", "Time from starting the request until the response headers arrive, including connection setup.", Unit.Ms, + [Stat.P50, Stat.P95, Stat.P99], good: 300, poor: 1000); + + public static readonly MetricDef Total = Metric.Histogram( + "http.total_ms", s_req, Ops.HttpPhaseMs("total_us"), "Total time", "Time until the response body was read.", Unit.Ms, [Stat.P50, Stat.P95, Stat.P99], good: 500, poor: 2000); + + public static readonly MetricDef FailTimeout = Metric.Counter( + "http.fail.timeout", s_req, Ops.HttpOutcomeCount(HttpOutcome.Timeout), "HTTP timeouts", "Requests that timed out.", new MetricExport("http.fail.timeout", Stat.Count, Unit.Count)); + + public static readonly MetricDef FailDns = Metric.Counter( + "http.fail.dns", s_req, Ops.HttpOutcomeCount(HttpOutcome.Dns), "DNS failures", "Requests whose host name did not resolve.", new MetricExport("http.fail.dns", Stat.Count, Unit.Count)); + + public static readonly MetricDef FailReset = Metric.Counter( + "http.fail.rst", s_req, Ops.HttpOutcomeCount(HttpOutcome.Reset), "HTTP resets", "Requests answered with a reset or refused.", new MetricExport("http.fail.rst", Stat.Count, Unit.Count)); + + public static readonly MetricDef FailTls = Metric.Counter( + "http.fail.tls", s_req, Ops.HttpOutcomeCount(HttpOutcome.Tls), "TLS failures", "Requests that ended in a TLS error.", new MetricExport("http.fail.tls", Stat.Count, Unit.Count)); + + public static readonly MetricDef FailStatus = Metric.Counter( + "http.fail.status", s_req, Ops.HttpOutcomeCount(HttpOutcome.BadStatus), "Bad status", "Requests that got an answer with an unaccepted status.", new MetricExport("http.fail.status", Stat.Count, Unit.Count)); + + public static readonly MetricDef FailUnsupported = Metric.Counter( + "http.fail.unsupported", s_req, Ops.HttpOutcomeCount(HttpOutcome.Unsupported), "Unsupported version", "Requests that could not use the requested HTTP version (for HTTP/3: no QUIC library, or the server does not offer it).", new MetricExport("http.fail.unsupported", Stat.Count, Unit.Count)); + + public static readonly MetricDef FailOther = Metric.Counter( + "http.fail.other", s_req, Ops.HttpOutcomeCount(HttpOutcome.Other), "Other failures", "Requests that failed for a reason not listed separately.", new MetricExport("http.fail.other", Stat.Count, Unit.Count)); + + public static readonly MetricDef Samples = Metric.Counter( + "n.http", s_req, Ops.RowCount, "http samples", "Number of http requests.", new MetricExport("n.http", Stat.Count, Unit.Count), Better.None); +} diff --git a/backend/src/MoShiro.Backend.Analysis/Metrics/LinkMetrics.cs b/backend/src/MoShiro.Backend.Analysis/Metrics/LinkMetrics.cs index 403b2bf..3f4908b 100644 --- a/backend/src/MoShiro.Backend.Analysis/Metrics/LinkMetrics.cs +++ b/backend/src/MoShiro.Backend.Analysis/Metrics/LinkMetrics.cs @@ -1,10 +1,10 @@ namespace MoShiro.Backend.Analysis; -/// 线路基础指标（udp_echo / icmp_echo 的回显流）。新增指标 = 加一行声明。 +/// 线路基础指标（quic_ping / icmp_ping 的回显流）。新增指标 = 加一行声明。 public static class LinkMetrics { - private static readonly string[] s_udpMsg = ["udp_echo.msg"]; - private static readonly string[] s_udpOrIcmpMsg = ["udp_echo.msg", "icmp_echo.msg"]; + private static readonly string[] s_udpMsg = ["quic_ping.msg"]; + private static readonly string[] s_udpOrIcmpMsg = ["quic_ping.msg", "icmp_ping.msg"]; public static readonly MetricDef Rtt = Metric.Histogram( "rtt", s_udpOrIcmpMsg, Ops.Rtt, "RTT", "Round-trip time with the Agent processing time removed.", Unit.Ms, @@ -45,5 +45,5 @@ public static class LinkMetrics "dup", s_udpMsg, Ops.Dup, "Duplicates", "Replies received more than once.", new MetricExport("dup", Stat.Count, Unit.Count)); public static readonly MetricDef SamplesUdp = Metric.Counter( - "n.udp_echo", s_udpMsg, Ops.RowCount, "udp_echo samples", "Number of udp_echo records.", new MetricExport("n.udp_echo", Stat.Count, Unit.Count), Better.None); + "n.quic_ping", s_udpMsg, Ops.RowCount, "quic_ping samples", "Number of quic_ping records.", new MetricExport("n.quic_ping", Stat.Count, Unit.Count), Better.None); } diff --git a/backend/src/MoShiro.Backend.Analysis/Metrics/MetricRegistry.cs b/backend/src/MoShiro.Backend.Analysis/Metrics/MetricRegistry.cs index a0d3066..98fa9df 100644 --- a/backend/src/MoShiro.Backend.Analysis/Metrics/MetricRegistry.cs +++ b/backend/src/MoShiro.Backend.Analysis/Metrics/MetricRegistry.cs @@ -15,6 +15,8 @@ public static class MetricRegistry TcpMetrics.Rtt, TcpMetrics.Slow, TcpMetrics.Stall, TcpMetrics.StallMax, TcpMetrics.Retrans, TcpMetrics.RetransUp, TcpMetrics.RetransDown, TcpMetrics.Srtt, TcpMetrics.BulkRate, TcpMetrics.BulkPeak, TcpMetrics.SamplesTcp, SessionMetrics.ConnectSuccess, SessionMetrics.ConnectTcp, SessionMetrics.ConnectTls, SessionMetrics.FailTimeout, SessionMetrics.FailReset, SessionMetrics.FailTls, + HttpMetrics.Success, HttpMetrics.Dns, HttpMetrics.Tcp, HttpMetrics.Ttfb, HttpMetrics.Total, + HttpMetrics.FailTimeout, HttpMetrics.FailDns, HttpMetrics.FailReset, HttpMetrics.FailTls, HttpMetrics.FailStatus, HttpMetrics.FailUnsupported, HttpMetrics.FailOther, HttpMetrics.Samples, SessionMetrics.HoldResets, SessionMetrics.HoldLifetime, SessionMetrics.TrainLoss, SessionMetrics.TrainJitter, SessionMetrics.TrainRate, SessionMetrics.SamplesConnect, ]; @@ -26,6 +28,7 @@ public static class MetricRegistry new("udp", "QUIC Datagram (legacy alias)", ["loss.up", "loss.down", "loss.total", "rtt.p50", "rtt.p95", "rtt.p99", "jitter.up.p95", "jitter.down.p95", "loss.run_max", "reorder", "dup"]), new("tcp", "TCP", ["tcp.rtt.p50", "tcp.rtt.p95", "tcp.rtt.p99", "tcp.rtt.p999", "tcp.rtt.max", "tcp.rtt.tail", "tcp.slow", "tcp.stall.per_min", "tcp.stall_max", "tcp.retrans", "tcp.retrans.up", "tcp.retrans.down", "tcp.srtt.p95"]), new("connect", "Connect", ["connect.success", "connect.tcp_ms.p95", "connect.tls_ms.p95", "connect.fail.timeout", "connect.fail.rst", "connect.fail.tls"]), + new("http", "HTTP", ["http.success", "http.dns_ms.p95", "http.tcp_ms.p95", "http.ttfb_ms.p50", "http.ttfb_ms.p95", "http.ttfb_ms.p99", "http.total_ms.p50", "http.total_ms.p95", "http.total_ms.p99", "http.fail.timeout", "http.fail.dns", "http.fail.rst", "http.fail.tls", "http.fail.status", "http.fail.unsupported", "http.fail.other"]), new("session", "Session", ["hold.resets.per_h", "hold.lifetime_max"]), new("scores", "Scores", ["score.base", "score.mc", "score.hy2", "score.reality"]), ]; diff --git a/backend/src/MoShiro.Backend.Analysis/Metrics/SessionMetrics.cs b/backend/src/MoShiro.Backend.Analysis/Metrics/SessionMetrics.cs index 66abca1..ce7ca2b 100644 --- a/backend/src/MoShiro.Backend.Analysis/Metrics/SessionMetrics.cs +++ b/backend/src/MoShiro.Backend.Analysis/Metrics/SessionMetrics.cs @@ -1,11 +1,11 @@ namespace MoShiro.Backend.Analysis; -/// TCP 建连、长连接、QUIC Datagram 压测（legacy kind udp_train）的指标。 +/// TCP 建连、长连接、QUIC Datagram 压测（legacy kind quic_stream）的指标。 public static class SessionMetrics { - private static readonly string[] s_conn = ["connect.conn"]; - private static readonly string[] s_hold = ["hold.hold"]; - private static readonly string[] s_train = ["udp_train.pkt"]; + private static readonly string[] s_conn = ["tcp_connect.conn"]; + private static readonly string[] s_hold = ["tls_hold.hold"]; + private static readonly string[] s_train = ["quic_stream.pkt"]; public static readonly MetricDef ConnectSuccess = new( "connect.success", AccKind.Ratio, s_conn, Ops.OutcomeRatio(ConnectOutcome.Ok), "Connect success", "Share of connection attempts that completed.", Better.Higher, diff --git a/backend/src/MoShiro.Backend.Analysis/Metrics/TcpMetrics.cs b/backend/src/MoShiro.Backend.Analysis/Metrics/TcpMetrics.cs index 1c27c3c..8b15708 100644 --- a/backend/src/MoShiro.Backend.Analysis/Metrics/TcpMetrics.cs +++ b/backend/src/MoShiro.Backend.Analysis/Metrics/TcpMetrics.cs @@ -3,9 +3,9 @@ namespace MoShiro.Backend.Analysis; /// TCP 指标：回显的 RTT 与卡顿、内核采样的重传、批量传输的吞吐。 public static class TcpMetrics { - private static readonly string[] s_tcpMsg = ["tcp_echo.msg"]; - private static readonly string[] s_tcpInfo = ["tcp_echo.tcp_info", "tcp_bulk.tcp_info"]; - private static readonly string[] s_bulkBytes = ["tcp_bulk.bytes"]; + private static readonly string[] s_tcpMsg = ["tcp_ping.msg"]; + private static readonly string[] s_tcpInfo = ["tcp_ping.tcp_info", "tcp_transfer.tcp_info"]; + private static readonly string[] s_bulkBytes = ["tcp_transfer.bytes"]; public static readonly MetricDef Rtt = Metric.Histogram( "tcp.rtt", s_tcpMsg, Ops.Rtt, "TCP RTT", "Application-level echo round trip over TCP (includes retransmission delay).", Unit.Ms, @@ -47,5 +47,5 @@ public static class TcpMetrics [Stat.Max], Better.Higher); public static readonly MetricDef SamplesTcp = Metric.Counter( - "n.tcp_echo", s_tcpMsg, Ops.RowCount, "tcp_echo samples", "Number of tcp_echo records.", new MetricExport("n.tcp_echo", Stat.Count, Unit.Count), Better.None); + "n.tcp_ping", s_tcpMsg, Ops.RowCount, "tcp_ping samples", "Number of tcp_ping records.", new MetricExport("n.tcp_ping", Stat.Count, Unit.Count), Better.None); } diff --git a/backend/src/MoShiro.Backend.Analysis/Models/Models.cs b/backend/src/MoShiro.Backend.Analysis/Models/Models.cs index 383d470..425562a 100644 --- a/backend/src/MoShiro.Backend.Analysis/Models/Models.cs +++ b/backend/src/MoShiro.Backend.Analysis/Models/Models.cs @@ -32,7 +32,7 @@ public sealed class BaseModel : ServiceModelBase /// /// Minecraft：TCP 上不会丢包，丢包表现为重传和卡顿（队头阻塞）。 -/// 没有 tcp_echo 数据时，用 udp_echo 的丢包与 RTT 代入简化的 TCP 恢复模型估算，并标记为 estimated。 +/// 没有 tcp_ping 数据时，用 quic_ping 的丢包与 RTT 代入简化的 TCP 恢复模型估算，并标记为 estimated。 /// public sealed class McModel : ServiceModelBase { @@ -99,7 +99,7 @@ public sealed class McModel : ServiceModelBase /// /// Hysteria2（Brutal 固定码率）：有 hy2_ladder 测试结果时，用可持续码率、有效吞吐和限速开始的位置评分； -/// 没有测试时，只能用持续的 udp_echo 趋势粗略估算，并标记为 estimated。 +/// 没有测试时，只能用持续的 quic_ping 趋势粗略估算，并标记为 estimated。 /// public sealed class Hy2Model : ServiceModelBase { diff --git a/backend/src/MoShiro.Backend.Analysis/Ops/HttpOps.cs b/backend/src/MoShiro.Backend.Analysis/Ops/HttpOps.cs new file mode 100644 index 0000000..7af5a80 --- /dev/null +++ b/backend/src/MoShiro.Backend.Analysis/Ops/HttpOps.cs @@ -0,0 +1,119 @@ +using MoShiro.Protocol; + +namespace MoShiro.Backend.Analysis; + +/// http 请求的结果，编码与 Agent 的 HttpOutcomes 一致。 +public enum HttpOutcome +{ + Ok = 0, + Timeout = 1, + Reset = 2, + Tls = 3, + Other = 4, + Dns = 5, + BadStatus = 6, + Unsupported = 7, + NotAllowed = 8, +} + +/// http 流（req）的公共读法：权重 w 是这一行代表的网格间隔数（自适应降频后每次请求代表更长的时间）。 +internal readonly struct HttpColumns +{ + public HttpColumns(RawStream stream) + { + Stream = stream; + Ts = stream.Find("ts"); + Outcome = stream.Find("outcome"); + Weights = stream.Find("w"); + } + + public RawStream Stream { get; } + + public RawColumn? Ts { get; } + + public RawColumn? Outcome { get; } + + private RawColumn? Weights { get; } + + public bool Has => Ts is not null && Outcome is not null; + + public int Weight(int row) => Weights is not null && Weights.IsValid(row) ? (int)Math.Clamp(Weights.Values[row], 1, 65535) : 1; +} + +/// 成功率（分子是按权重累计的成功次数）或某种结果的次数。 +internal sealed class HttpOutcomeOp : MetricOp +{ + private readonly HttpOutcome _code; + private readonly bool _asRatio; + + public HttpOutcomeOp(HttpOutcome code, bool asRatio) + { + _code = code; + _asRatio = asRatio; + } + + public override string Signature => $"http_outcome/{_code}/{(_asRatio ? "ratio" : "count")}/1"; + + public override void Evaluate(ChunkView view, RawStream stream, IObservationSink sink) + { + var c = new HttpColumns(stream); + if (!c.Has) + { + return; + } + + for (int i = 0; i < stream.Count; i++) + { + bool hit = c.Outcome!.Values[i] == (long)_code; + long at = view.ColumnToUtcMs(c.Ts!, c.Ts!.Values[i]); + if (_asRatio) + { + int w = c.Weight(i); + sink.Emit(at, hit ? w : 0, w); + } + else if (hit) + { + sink.Emit(at, 1); + } + } + } +} + +/// 一个阶段的耗时（毫秒）：只统计成功的请求，按权重累计，所以分位数是按时间平均的。 +internal sealed class HttpPhaseMsOp : MetricOp +{ + private const int MaxRepeat = 256; + + private readonly string _column; + + public HttpPhaseMsOp(string column) + { + _column = column; + } + + public override string Signature => $"http_phase_ms/{_column}/1"; + + public override void Evaluate(ChunkView view, RawStream stream, IObservationSink sink) + { + var c = new HttpColumns(stream); + RawColumn? column = stream.Find(_column); + if (!c.Has || column is null) + { + return; + } + + for (int i = 0; i < stream.Count; i++) + { + if (c.Outcome!.Values[i] != (long)HttpOutcome.Ok || !column.IsValid(i)) + { + continue; + } + + long at = view.ColumnToUtcMs(c.Ts!, c.Ts!.Values[i]); + for (int k = Math.Min(c.Weight(i), MaxRepeat); k > 0; k--) + { + sink.Emit(at, column.Values[i] / 1000.0); + } + } + } +} diff --git a/backend/src/MoShiro.Backend.Analysis/Ops/Ops.cs b/backend/src/MoShiro.Backend.Analysis/Ops/Ops.cs index 0713448..dde5aac 100644 --- a/backend/src/MoShiro.Backend.Analysis/Ops/Ops.cs +++ b/backend/src/MoShiro.Backend.Analysis/Ops/Ops.cs @@ -37,6 +37,12 @@ public static class Ops public static MetricOp OutcomeCount(ConnectOutcome code) => new OutcomeOp(code, asRatio: false); + public static MetricOp HttpSuccess { get; } = new HttpOutcomeOp(HttpOutcome.Ok, asRatio: true); + + public static MetricOp HttpOutcomeCount(HttpOutcome code) => new HttpOutcomeOp(code, asRatio: false); + + public static MetricOp HttpPhaseMs(string column) => new HttpPhaseMsOp(column); + public static MetricOp HoldEvents(HoldEvent holdEvent) => new HoldEventOp(holdEvent); public static MetricOp HoldLifetime { get; } = new HoldLifetimeOp(); diff --git a/backend/src/MoShiro.Backend.Analysis/Ops/SessionOps.cs b/backend/src/MoShiro.Backend.Analysis/Ops/SessionOps.cs index 77c379b..ccf0d88 100644 --- a/backend/src/MoShiro.Backend.Analysis/Ops/SessionOps.cs +++ b/backend/src/MoShiro.Backend.Analysis/Ops/SessionOps.cs @@ -330,7 +330,7 @@ internal sealed class HoldLifetimeOp : MetricOp } } -/// 单向 UDP 流（udp_train）的丢包：接收端没有记录到 t_recv 即为丢失。 +/// 单向 UDP 流（quic_stream）的丢包：接收端没有记录到 t_recv 即为丢失。 internal sealed class TrainLossOp : MetricOp { public override string Signature => "train_loss/1"; diff --git a/backend/src/MoShiro.Backend.Api/ApiModule.cs b/backend/src/MoShiro.Backend.Api/ApiModule.cs index 552b49b..6e6ddf0 100644 --- a/backend/src/MoShiro.Backend.Api/ApiModule.cs +++ b/backend/src/MoShiro.Backend.Api/ApiModule.cs @@ -98,6 +98,7 @@ public static class ApiModule { RouteGroupBuilder publicGroup = app.MapGroup("/api/public").AddEndpointFilter(PublicGuard.InvokeAsync); QueryEndpoints.Map(publicGroup); + TargetEndpoints.Map(publicGroup); LiveEndpoints.Map(publicGroup); HostEndpoints.Map(publicGroup); NodeEndpoints.Map(publicGroup); diff --git a/backend/src/MoShiro.Backend.Api/Contracts/ApiJsonContext.cs b/backend/src/MoShiro.Backend.Api/Contracts/ApiJsonContext.cs index 0967ec6..d56d60f 100644 --- a/backend/src/MoShiro.Backend.Api/Contracts/ApiJsonContext.cs +++ b/backend/src/MoShiro.Backend.Api/Contracts/ApiJsonContext.cs @@ -25,6 +25,7 @@ namespace MoShiro.Backend.Api; [JsonSerializable(typeof(QuicLineHealthDto))] [JsonSerializable(typeof(TestDto))] [JsonSerializable(typeof(PageDto))] +[JsonSerializable(typeof(PageDto))] [JsonSerializable(typeof(PageDto))] [JsonSerializable(typeof(PageDto))] [JsonSerializable(typeof(PageDto))] diff --git a/backend/src/MoShiro.Backend.Api/Contracts/PublicDtos.cs b/backend/src/MoShiro.Backend.Api/Contracts/PublicDtos.cs index 4994440..fe9226a 100644 --- a/backend/src/MoShiro.Backend.Api/Contracts/PublicDtos.cs +++ b/backend/src/MoShiro.Backend.Api/Contracts/PublicDtos.cs @@ -219,3 +219,24 @@ public sealed record LiveHostDto(string Id, bool Online, HostDto Host); /// 主机资源占用的历史（列式，按步长取平均）。 public sealed record HostSeriesDto(string Id, DateTimeOffset From, DateTimeOffset To, string Step, IReadOnlyList T, IReadOnlyDictionary Values); + +/// +/// 一个 Agent 上的一个单个 Agent 任务：它测的目标、现在的状态和最近一段时间的时序。 +/// status：running 有最近的数据；waiting 任务已下发、还没有数据；stale 有过数据但最近没有； +/// rejected Agent 本机拒绝执行（reason 是原因）；offline Agent 不在线。 +/// +public sealed record TargetDto( + string TaskId, + string AgentId, + string AgentName, + string Kind, + string Target, + string Status, + string? Reason, + DateTimeOffset? LastSample, + DateTimeOffset From, + DateTimeOffset To, + string Step, + IReadOnlyList T, + IReadOnlyDictionary Values, + SummaryDto Summary); diff --git a/backend/src/MoShiro.Backend.Api/Public/TargetEndpoints.cs b/backend/src/MoShiro.Backend.Api/Public/TargetEndpoints.cs new file mode 100644 index 0000000..b95ccf5 --- /dev/null +++ b/backend/src/MoShiro.Backend.Api/Public/TargetEndpoints.cs @@ -0,0 +1,112 @@ +using System.Text.Json; +using Fleet.Core; +using Microsoft.AspNetCore.Builder; +using Microsoft.AspNetCore.Http; +using Microsoft.AspNetCore.Routing; +using MoShiro.Backend.Application; +using MoShiro.Protocol; + +namespace MoShiro.Backend.Api; + +/// 单个 Agent 任务（Agent 自己去测一个外部地址）：每个 Agent 实际在执行的目标、状态和最近的时序。 +public static class TargetEndpoints +{ + private const string Tag = "public"; + + /// 最近多久内有样本算「在运行」：自适应退避到最慢是几分钟一次，所以要留够余量。 + private static readonly TimeSpan s_running = TimeSpan.FromMinutes(15); + + /// 每种方法默认看的指标。 + private static readonly Dictionary s_metrics = new(StringComparer.Ordinal) + { + ["http_request"] = "http.success,http.ttfb_ms.p50,http.ttfb_ms.p95,http.total_ms.p95,http.dns_ms.p95,http.tcp_ms.p95,http.fail.timeout,http.fail.dns,http.fail.rst,http.fail.tls,http.fail.status,http.fail.unsupported,http.fail.other", + ["tcp_connect"] = "connect.success,connect.tcp_ms.p95,connect.tls_ms.p95", + ["icmp_ping"] = "rtt.p50,rtt.p95,loss.total", + }; + + /// 详情页要的指标：卡片上的那些再加上分位数和各种失败计数。 + private static readonly Dictionary s_fullMetrics = new(StringComparer.Ordinal) + { + ["http_request"] = "http.success,http.ttfb_ms.p50,http.ttfb_ms.p95,http.ttfb_ms.p99,http.total_ms.p50,http.total_ms.p95,http.total_ms.p99,http.dns_ms.p95,http.tcp_ms.p95,http.fail.timeout,http.fail.dns,http.fail.rst,http.fail.tls,http.fail.status,http.fail.unsupported,http.fail.other,n.http", + ["tcp_connect"] = "connect.success,connect.tcp_ms.p95,connect.tls_ms.p95,connect.fail.timeout,connect.fail.rst,connect.fail.tls,n.connect", + ["icmp_ping"] = "rtt.p50,rtt.p95,rtt.p99,loss.total", + }; + + public static void Map(RouteGroupBuilder group) => + group.MapGet("/targets", ListAsync) + .Api("targets_list", "Single-Agent tasks with their status and recent series", Tag) + .Returns>().ProducesProblems(403, 422, 429, 503); + + /// agent_id 和 task_id 只看一个目标；full=true 返回详情页要的全部指标。 + private static Task ListAsync( + HttpContext http, TargetService targets, QueryService query, BackendOptions options, TimeProvider time, + string? range, DateTimeOffset? from, DateTimeOffset? to, string? step, string? agent_id, string? task_id, bool? full) => + QueryHelpers.GuardedAsync(http, async ct => + { + IReadOnlyList views = await targets.ListAsync(a => a.Agent.PublicVisible || PublicGuard.CanSeeHidden(http), ct); + var items = new List(views.Count); + long now = time.GetUtcNow().ToUnixTimeMilliseconds(); + foreach (TargetView view in views.Where(v => (agent_id is null || v.Agent.Agent.Id.Value == agent_id) && (task_id is null || v.Task.Id.Value == task_id))) + { + string wanted = (full == true ? s_fullMetrics : s_metrics).GetValueOrDefault(view.Task.Kind, "rtt.p50"); + Result selection = MetricSelection.Resolve(wanted, null, options.MaxMetricsPerQuery); + if (!selection.TryGet(out MetricSelection? sel, out FleetError? error)) + { + return Problems.From(http, error); + } + + Result series = await query.TargetSeriesAsync(view.Agent.Agent.Id, view.Task.Id, new RangeRequest(range ?? "3h", from, to, step, null, null), sel, ct); + if (!series.TryGet(out TargetSeriesResult? result, out error)) + { + return Problems.From(http, error); + } + + long? latest = await query.TargetLatestAsync(view.Agent.Agent.Id, view.Task.Id, ct); + (string status, string? reason) = StatusOf(view, latest, now); + items.Add(new TargetDto( + view.Task.Id.Value, view.Agent.Agent.Id.Value, view.Agent.Agent.Name, view.Task.Kind, view.Task.Target ?? "", status, reason, + latest is { } at ? Mappers.Time(at) : null, Mappers.Time(result.Range.FromMs), Mappers.Time(result.Range.ToMs), Steps.Format(result.Range.StepMs), + result.T, result.Values, Mappers.ToDto(result.Summary))); + } + + return TypedResults.Json(new PageDto(items, null), ApiJsonContext.Default.PageDtoTargetDto); + }); + + private static (string Status, string? Reason) StatusOf(TargetView view, long? latest, long now) + { + if (RejectedReason(view) is { } reason) + { + return ("rejected", reason); + } + + if (!view.Agent.Online) + { + return ("offline", null); + } + + if (latest is null) + { + return ("waiting", null); + } + + return (now - latest.Value <= s_running.TotalMilliseconds ? "running" : "stale", null); + } + + /// Agent 本机拒绝执行这个任务的原因（心跳里的 rejected[]，条目 id 是任务 id）。 + private static string? RejectedReason(TargetView view) + { + if (view.Agent.Agent.RejectedJson is not { } json) + { + return null; + } + + try + { + return JsonSerializer.Deserialize(json, ApiJsonContext.Default.ListRejectedItem)?.FirstOrDefault(r => r.Id == view.Task.Id.Value)?.Reason; + } + catch (JsonException) + { + return null; + } + } +} diff --git a/backend/src/MoShiro.Backend.Application/ApplicationServiceCollectionExtensions.cs b/backend/src/MoShiro.Backend.Application/ApplicationServiceCollectionExtensions.cs index b53ef08..bc32064 100644 --- a/backend/src/MoShiro.Backend.Application/ApplicationServiceCollectionExtensions.cs +++ b/backend/src/MoShiro.Backend.Application/ApplicationServiceCollectionExtensions.cs @@ -32,6 +32,7 @@ public static class ApplicationServiceCollectionExtensions services.AddSingleton(); services.AddSingleton(); services.AddSingleton(); + services.AddSingleton(); services.AddSingleton(); services.AddSingleton(); services.AddSingleton(); diff --git a/backend/src/MoShiro.Backend.Application/Query/QueryResults.cs b/backend/src/MoShiro.Backend.Application/Query/QueryResults.cs index 6392195..1b020e8 100644 --- a/backend/src/MoShiro.Backend.Application/Query/QueryResults.cs +++ b/backend/src/MoShiro.Backend.Application/Query/QueryResults.cs @@ -45,3 +45,12 @@ public sealed record CompareResult( string Metric, QueryRange Range, SourcePlan Plan); + +/// 单个 Agent 任务（目标任务）的时序：和线路时序同一套分层与汇总，只是没有线路、事件和评分。 +public sealed record TargetSeriesResult( + QueryRange Range, + SourcePlan Plan, + IReadOnlyList T, + IReadOnlyDictionary Values, + SummaryResult Summary, + IReadOnlyDictionary Versions); diff --git a/backend/src/MoShiro.Backend.Application/Query/QueryService.cs b/backend/src/MoShiro.Backend.Application/Query/QueryService.cs index 0bf4a61..47d0af4 100644 --- a/backend/src/MoShiro.Backend.Application/Query/QueryService.cs +++ b/backend/src/MoShiro.Backend.Application/Query/QueryService.cs @@ -138,6 +138,61 @@ public sealed class QueryService return Result.Ok(new SeriesResult(line, p.Range, p.Plan, bucketed.T, values, summary, gaps, events, versions)); } + /// 某个 Agent 上一个单个 Agent 任务的时序（序列 target:<任务 id>）。 + public async ValueTask> TargetSeriesAsync(NodeId agent, TaskId task, RangeRequest request, MetricSelection selection, CancellationToken ct) + { + Result<(QueryRange Range, SourcePlan Plan)> planned = Plan(request, "3h"); + if (!planned.TryGet(out (QueryRange Range, SourcePlan Plan) p, out FleetError? error)) + { + return Result.Fail(error); + } + + Result> loaded = await _loader.LoadAsync(p.Plan, p.Range, agent, TargetSeries(task), ct); + if (!loaded.TryGet(out IReadOnlyList? rows, out error)) + { + return Result.Fail(error); + } + + BucketedRows bucketed = BucketedRows.Create(rows, p.Range); + int n = p.Range.Points; + var values = selection.ExportIds.ToDictionary(id => id, _ => new double?[n], StringComparer.Ordinal); + double widthSeconds = p.Plan.LevelWidthMs / 1000.0; + for (int i = 0; i < n; i++) + { + if (bucketed.Sets[i] is not { } set) + { + continue; + } + + Dictionary exported = ExportEvaluator.Evaluate(set, selection.ExportIds, bucketed.RowCounts[i] * widthSeconds); + foreach (string id in selection.ExportIds) + { + if (exported.TryGetValue(id, out double v)) + { + values[id][i] = v; + } + } + } + + var versions = new Dictionary(StringComparer.Ordinal); + foreach (string id in selection.ExportIds) + { + if (MetricRegistry.TryFindExport(id, out MetricDef def, out _)) + { + versions[id] = def.Fingerprint; + } + } + + SummaryResult summary = Summarize(bucketed.Merge(), bucketed.TotalRows, p.Plan, p.Range, selection, new Dictionary()); + return Result.Ok(new TargetSeriesResult(p.Range, p.Plan, bucketed.T, values, summary, versions)); + } + + /// 这个单个 Agent 任务最后一个样本的时刻（Unix 毫秒）；还没有数据返回 null。 + public ValueTask TargetLatestAsync(NodeId agent, TaskId task, CancellationToken ct) => + _loader.LatestSampleMsAsync(agent, TargetSeries(task), _time.GetUtcNow().ToUnixTimeMilliseconds(), ct); + + private static string TargetSeries(TaskId task) => "target:" + task.Value; + /// 这条线路最早的数据（Unix 毫秒）；没有数据返回 null。 public ValueTask EarliestAsync(LineKey line, CancellationToken ct) => _loader.EarliestSampleMsAsync(line.Listener, line.Series, _time.GetUtcNow().ToUnixTimeMilliseconds(), ct); diff --git a/backend/src/MoShiro.Backend.Application/Query/Targets.cs b/backend/src/MoShiro.Backend.Application/Query/Targets.cs new file mode 100644 index 0000000..ca193c3 --- /dev/null +++ b/backend/src/MoShiro.Backend.Application/Query/Targets.cs @@ -0,0 +1,39 @@ +using Fleet.Server; + +namespace MoShiro.Backend.Application; + +/// 一个 Agent 上正在（或应该）运行的单个 Agent 任务：这个 Agent 的有效任务集里 scope=target 的那些。 +public sealed record TargetView(AgentView Agent, EffectiveTask Task); + +/// 单个 Agent 任务目录：按 Agent 展开模板、覆盖和专属任务，列出每个 Agent 实际执行的目标。 +public sealed class TargetService +{ + private readonly AgentService _agents; + private readonly ITaskStore _tasks; + + public TargetService(AgentService agents, ITaskStore tasks) + { + _agents = agents; + _tasks = tasks; + } + + public async ValueTask> ListAsync(Func include, CancellationToken ct) + { + IReadOnlyList all = await _tasks.ListAsync(ct); + var result = new List(); + foreach (AgentView view in await _agents.ListAsync(ct)) + { + if (view.Agent.Status != NodeStatus.Active || !include(view)) + { + continue; + } + + foreach (EffectiveTask task in EffectiveTasks.Compute(all, view.Agent, filterByAgentKinds: false).Where(t => t.Scope == EffectiveTasks.TargetScope)) + { + result.Add(new TargetView(view, task)); + } + } + + return result; + } +} diff --git a/backend/src/MoShiro.Backend.Application/Tasks/EffectiveTasks.cs b/backend/src/MoShiro.Backend.Application/Tasks/EffectiveTasks.cs index ae996ef..1fe16b3 100644 --- a/backend/src/MoShiro.Backend.Application/Tasks/EffectiveTasks.cs +++ b/backend/src/MoShiro.Backend.Application/Tasks/EffectiveTasks.cs @@ -13,7 +13,7 @@ public sealed record EffectiveTask( ParamValues Resolved); /// -/// 有效任务集：线路任务模板 + 针对单个 Agent 的覆盖 + 该 Agent 的目标任务。 +/// 有效任务集：任务模板（线路任务、单个 Agent 的任务）+ 针对单个 Agent 的覆盖 + 只属于该 Agent 的目标任务。 /// 参数经 ParamSpec 校验并补全默认值，Backend 和节点使用同一套规则。 /// public static class EffectiveTasks @@ -35,23 +35,23 @@ public static class EffectiveTasks foreach (TaskRecord task in all.OrderBy(t => t.Id.Value, StringComparer.Ordinal)) { - if (task.Scope == LineScope && task.AgentId is null) + if (task.AgentId is null) { - // 模板：有覆盖时以覆盖为准（覆盖可以停用或改参数） + // 模板（线路任务或单个 Agent 的任务）：有覆盖时以覆盖为准（覆盖可以停用、改参数，单个 Agent 的任务还可以换目标） TaskRecord? over = overrides.GetValueOrDefault(task.Id); if (over is not null) { if (over.Enabled) { - Add(result, over.Id, task.Scope, over.Kind, null, Merge(task.Params, over.Params), over.Schedule, agent, filterByAgentKinds); + Add(result, over.Id, task.Scope, over.Kind, over.Target ?? task.Target, Merge(task.Params, over.Params), over.Schedule, agent, filterByAgentKinds); } } else if (task.Enabled) { - Add(result, task.Id, task.Scope, task.Kind, null, task.Params, task.Schedule, agent, filterByAgentKinds); + Add(result, task.Id, task.Scope, task.Kind, task.Target, task.Params, task.Schedule, agent, filterByAgentKinds); } } - else if (task.Scope == TargetScope && task.AgentId == agent.Id && task.Enabled) + else if (task.Scope == TargetScope && task.AgentId == agent.Id && task.BaseTaskId is null && task.Enabled) { Add(result, task.Id, task.Scope, task.Kind, task.Target, task.Params, task.Schedule, agent, filterByAgentKinds); } diff --git a/backend/src/MoShiro.Backend.Application/Tasks/TaskService.cs b/backend/src/MoShiro.Backend.Application/Tasks/TaskService.cs index 50687fe..70fd4b8 100644 --- a/backend/src/MoShiro.Backend.Application/Tasks/TaskService.cs +++ b/backend/src/MoShiro.Backend.Application/Tasks/TaskService.cs @@ -41,7 +41,7 @@ public sealed class TaskService public ValueTask GetAsync(TaskId id, CancellationToken ct) => _tasks.GetAsync(id, ct); /// - /// 创建默认的线路任务模板，这样新 Agent 注册后自动获得任务：udp_echo、tcp_echo 持续运行，connect（TCP 建连和 TLS 握手耗时， + /// 创建默认的线路任务模板，这样新 Agent 注册后自动获得任务：quic_ping、tcp_ping 持续运行，connect（TCP 建连和 TLS 握手耗时， /// 每分钟 6 次）也持续运行。每一批默认任务只种一次，用设置里的标记记住；你删掉的默认模板不会被重新建出来， /// 后来新增的默认任务（connect）在已有部署升级后会补建一次。 /// @@ -49,13 +49,13 @@ public sealed class TaskService { if (await _settings.GetAsync("tasks.seeded", ct) is null) { - await SeedAsync([Kinds.UdpEcho, Kinds.TcpEcho], ct); + await SeedAsync([Kinds.QuicPing, Kinds.TcpPing], ct); await _settings.SetAsync("tasks.seeded", "1", ct); } if (await _settings.GetAsync("tasks.seeded.connect", ct) is null) { - await SeedAsync([Kinds.Connect], ct); + await SeedAsync([Kinds.TcpConnect], ct); await _settings.SetAsync("tasks.seeded.connect", "1", ct); } } @@ -223,9 +223,31 @@ public sealed class TaskService if (scope == EffectiveTasks.TargetScope) { - if (agentId is null || !TryNormalizeTarget(target, out string? normalized)) + // 单个 Agent 的任务：没有 agent_id 是模板（所有 Agent 各自对这个目标执行），有 agent_id 是只属于这个 Agent 的任务， + // 或者（带 base_task_id）覆盖模板的参数、停用它、换一个目标 + if (baseTaskId is { } templateId) { - return ValidationError.Fail("target", agentId is null ? "agent_id is required for target tasks" : "must be an IP address with an optional port"); + TaskRecord? template = await _tasks.GetAsync(templateId, ct); + if (agentId is null || template is null || template.AgentId is not null || template.Scope != EffectiveTasks.TargetScope) + { + return ValidationError.Fail("base_task_id", "must reference a single-agent template and requires agent_id"); + } + + if (template.Kind != kind) + { + return ValidationError.Fail("kind", "an override must use the same kind as its template"); + } + + if (target is null) + { + return Result.Ok(new ValidatedTask(null)); + } + } + + string? normalized = TargetHost.Normalize(kind, target); + if (normalized is null) + { + return ValidationError.Fail("target", kind == TargetHost.HttpKind ? "must be an http:// or https:// URL" : "must be a host name or IP address with an optional port"); } return Result.Ok(new ValidatedTask(normalized)); @@ -288,28 +310,4 @@ public sealed class TaskService return Result.Ok(true); } - - /// 只接受 IP 地址（可带端口），不做 DNS 解析，避免把 Agent 用作解析器。 - public static bool TryNormalizeTarget(string? target, out string? normalized) - { - normalized = null; - if (string.IsNullOrWhiteSpace(target)) - { - return false; - } - - if (IPAddress.TryParse(target, out IPAddress? ip)) - { - normalized = ip.ToString(); - return true; - } - - if (IPEndPoint.TryParse(target, out IPEndPoint? endpoint) && endpoint.Port is > 0 and <= 65535) - { - normalized = endpoint.ToString(); - return true; - } - - return false; - } } diff --git a/backend/src/MoShiro.Backend.Application/Tests/Plans.cs b/backend/src/MoShiro.Backend.Application/Tests/Plans.cs index f4c72da..50fb696 100644 --- a/backend/src/MoShiro.Backend.Application/Tests/Plans.cs +++ b/backend/src/MoShiro.Backend.Application/Tests/Plans.cs @@ -18,7 +18,7 @@ public static class Plans private static JsonElement S(string value) => JsonSerializer.SerializeToElement(value, PlanJson.Default.String); private static PhaseSpec Train(double mbps) => - new("udp_train", new Dictionary { ["mbps"] = N(mbps), ["duration_s"] = N(10) }, 10); + new("quic_stream", new Dictionary { ["mbps"] = N(mbps), ["duration_s"] = N(10) }, 10); public static readonly PlanSpec Hy2Ladder = new( "hy2_ladder", @@ -31,9 +31,9 @@ public static class Plans "REALITY session", "Connection establishment, a short (4 s) bulk transfer for the peak throughput and a held connection.", [ - new PhaseSpec("connect", new Dictionary { ["per_minute"] = N(20) }, 30), - new PhaseSpec("tcp_bulk", new Dictionary { ["duration_s"] = N(4), ["direction"] = S("rev") }, 4), - new PhaseSpec("hold", new Dictionary { ["heartbeat_s"] = N(5) }, 30), + new PhaseSpec("tcp_connect", new Dictionary { ["per_minute"] = N(20) }, 30), + new PhaseSpec("tcp_transfer", new Dictionary { ["duration_s"] = N(4), ["direction"] = S("rev") }, 4), + new PhaseSpec("tls_hold", new Dictionary { ["heartbeat_s"] = N(5) }, 30), ]); public static readonly IReadOnlyList All = [Hy2Ladder, RealitySession]; diff --git a/backend/src/MoShiro.Backend.Storage.Sqlite/Schema.cs b/backend/src/MoShiro.Backend.Storage.Sqlite/Schema.cs index 3c72cb4..1ec6b5b 100644 --- a/backend/src/MoShiro.Backend.Storage.Sqlite/Schema.cs +++ b/backend/src/MoShiro.Backend.Storage.Sqlite/Schema.cs @@ -3,12 +3,12 @@ using Microsoft.Data.Sqlite; namespace MoShiro.Backend.Storage.Sqlite; /// -/// 表结构。版本记录在 PRAGMA user_version。当前只有一个版本：不是这个版本的数据库（包括更早的构建留下的）不会被迁移， -/// 启动时报错，删除 moshiro.db 和 raw/ 之后重新开始。以后的结构变更按版本号追加内置 SQL。 +/// 表结构。版本记录在 PRAGMA user_version。版本 1 到 2 只改了 Kind 的名字（见 ）； +/// 其他版本的数据库（包括更早的构建留下的）不会被迁移，启动时报错，删除 moshiro.db 和 raw/ 之后重新开始。以后的结构变更按版本号追加内置 SQL。 /// internal static class Schema { - public const int CurrentVersion = 1; + public const int CurrentVersion = 2; public static void Migrate(SqliteConnection connection) { @@ -18,6 +18,21 @@ internal static class Schema return; } + if (version == 1) + { + using SqliteTransaction upgrade = connection.BeginTransaction(); + foreach ((string from, string to) in RenameKinds) + { + connection.Exec($"UPDATE tasks SET kind = '{to}' WHERE kind = '{from}'"); + connection.Exec($"UPDATE raw_chunks SET kind = '{to}' WHERE kind = '{from}'"); + connection.Exec($"UPDATE tests SET plan = REPLACE(plan, '\"{from}\"', '\"{to}\"')"); + } + + connection.Exec("PRAGMA user_version = 2"); + upgrade.Commit(); + return; + } + if (version != 0) { throw new InvalidOperationException($"database schema version {version} is not supported by this build (expected {CurrentVersion}); delete moshiro.db and the raw/ directory to start over"); @@ -25,10 +40,17 @@ internal static class Schema using SqliteTransaction transaction = connection.BeginTransaction(); connection.Exec(V1); - connection.Exec("PRAGMA user_version = 1"); + connection.Exec("PRAGMA user_version = 2"); transaction.Commit(); } + /// 版本 2：Kind 改成「协议 + 做什么」的名字。任务和原始数据索引跟着改；已经落盘的原始块头里还是旧名字，按旧名字解不出指标，重算不到它们，已经算好的指标不受影响。 + private static readonly (string From, string To)[] RenameKinds = + [ + ("udp_echo", "quic_ping"), ("tcp_echo", "tcp_ping"), ("udp_train", "quic_stream"), ("tcp_bulk", "tcp_transfer"), + ("connect", "tcp_connect"), ("hold", "tls_hold"), ("icmp_echo", "icmp_ping"), + ]; + /// /// 新增 Kind 或指标都不需要改表结构：指标是 metric_*.metrics 里的一个键，Kind 只是 raw_chunks.kind 的一个取值。 /// 节点只有一张表（agents）：监听的节点有地址和端口，不监听的节点只拨出。线路 (agent_id, series) 里的 agent_id 是监听方，series 里是拨号方。 diff --git a/backend/tests/MoShiro.Backend.Analysis.Tests/ModelAndTestEvaluatorTests.cs b/backend/tests/MoShiro.Backend.Analysis.Tests/ModelAndTestEvaluatorTests.cs index 42d667a..71da243 100644 --- a/backend/tests/MoShiro.Backend.Analysis.Tests/ModelAndTestEvaluatorTests.cs +++ b/backend/tests/MoShiro.Backend.Analysis.Tests/ModelAndTestEvaluatorTests.cs @@ -32,8 +32,8 @@ public sealed class ModelAndTestEvaluatorTests [Fact] public void Fingerprint_ChangesWhenTheOpChanges() { - var a = Metric.Counter("x", ["udp_echo.msg"], Ops.Gap(GapOutput.Count, "rate_hz", 100), "x", "x", new MetricExport("x", Stat.Count, Unit.Count)); - var b = Metric.Counter("x", ["udp_echo.msg"], Ops.Gap(GapOutput.Count, "rate_hz", 200), "x", "x", new MetricExport("x", Stat.Count, Unit.Count)); + var a = Metric.Counter("x", ["quic_ping.msg"], Ops.Gap(GapOutput.Count, "rate_hz", 100), "x", "x", new MetricExport("x", Stat.Count, Unit.Count)); + var b = Metric.Counter("x", ["quic_ping.msg"], Ops.Gap(GapOutput.Count, "rate_hz", 200), "x", "x", new MetricExport("x", Stat.Count, Unit.Count)); Assert.NotEqual(a.Fingerprint, b.Fingerprint); } diff --git a/backend/tests/MoShiro.Backend.Analysis.Tests/OpTests.cs b/backend/tests/MoShiro.Backend.Analysis.Tests/OpTests.cs index dfbe7e4..ce7e66c 100644 --- a/backend/tests/MoShiro.Backend.Analysis.Tests/OpTests.cs +++ b/backend/tests/MoShiro.Backend.Analysis.Tests/OpTests.cs @@ -25,7 +25,7 @@ public sealed class OpTests [Fact] public void Rtt_RemovesAgentProcessingTime() { - RawChunk chunk = ChunkFactory.Echo(s_agent, "udp_echo", T0, 100, 100, _ => new EchoPacket(false, false, 20_000, 30_000, ProcessUs: 5_000)); + RawChunk chunk = ChunkFactory.Echo(s_agent, "quic_ping", T0, 100, 100, _ => new EchoPacket(false, false, 20_000, 30_000, ProcessUs: 5_000)); var rtt = (HistogramAcc)EvaluateOneBucket(chunk, "rtt").Items["rtt"]; Assert.Equal(100UL, rtt.Count); Assert.InRange(rtt.Quantile(0.5)!.Value, 50 * 0.955, 50 * 1.045); @@ -37,8 +37,8 @@ public sealed class OpTests static EchoPacket At(int i) => new(LostUp: i % 20 < 2, LostDown: false, i % 10 < 2 ? 100_000 : 20_000, 20_000); // 每 100 ms 一个包，每个权重 1；对照：只取偶数行，每 200 ms 一个包，每个权重 2——代表同样长的时间 - RawChunk dense = ChunkFactory.Echo(s_agent, "udp_echo", T0, 100, 100, At); - RawChunk sparse = ChunkFactory.Echo(s_agent, "udp_echo", T0, 50, 200, i => At(i * 2), weight: _ => 2); + RawChunk dense = ChunkFactory.Echo(s_agent, "quic_ping", T0, 100, 100, At); + RawChunk sparse = ChunkFactory.Echo(s_agent, "quic_ping", T0, 50, 200, i => At(i * 2), weight: _ => 2); MetricSet d = EvaluateOneBucket(dense, "rtt", "loss.up", "loss.total"); MetricSet s = EvaluateOneBucket(sparse, "rtt", "loss.up", "loss.total"); @@ -54,7 +54,7 @@ public sealed class OpTests public void Loss_SeparatesUpAndDown() { RawChunk chunk = ChunkFactory.Echo( - s_agent, "udp_echo", T0, 100, 100, + s_agent, "quic_ping", T0, 100, 100, i => new EchoPacket(LostUp: i < 10, LostDown: i is >= 10 and < 15, 20_000, 20_000)); MetricSet set = EvaluateOneBucket(chunk, "loss.up", "loss.down", "loss.total"); @@ -77,7 +77,7 @@ public sealed class OpTests public void Jitter_OnlyOneDirectionVaries_IsAttributedToThatDirection() { // 下行时延每隔一个包多 20 ms - RawChunk chunk = ChunkFactory.Echo(s_agent, "udp_echo", T0, 100, 100, i => new EchoPacket(false, false, 20_000, 20_000 + (i % 2 == 0 ? 20_000 : 0))); + RawChunk chunk = ChunkFactory.Echo(s_agent, "quic_ping", T0, 100, 100, i => new EchoPacket(false, false, 20_000, 20_000 + (i % 2 == 0 ? 20_000 : 0))); MetricSet set = EvaluateOneBucket(chunk, "jitter.up", "jitter.down"); Assert.True(((HistogramAcc)set.Items["jitter.up"]).Quantile(0.95) < 0.5); Assert.InRange(((HistogramAcc)set.Items["jitter.down"]).Quantile(0.95)!.Value, 15, 25); @@ -95,7 +95,7 @@ public sealed class OpTests public void LossRuns_ReportsConsecutiveLosses() { bool[] lost = [false, true, true, true, false, true, false, false, true, true]; - RawChunk chunk = ChunkFactory.Echo(s_agent, "udp_echo", T0, lost.Length, 100, i => new EchoPacket(false, lost[i], 20_000, 20_000)); + RawChunk chunk = ChunkFactory.Echo(s_agent, "quic_ping", T0, lost.Length, 100, i => new EchoPacket(false, lost[i], 20_000, 20_000)); var runs = (RunLengthAcc)EvaluateOneBucket(chunk, "loss.run").Items["loss.run"]; Assert.Equal(3UL, runs.Runs); Assert.Equal(3, runs.Max); @@ -105,7 +105,7 @@ public sealed class OpTests [Fact] public void LossRuns_TrailingRunIsReported() { - RawChunk chunk = ChunkFactory.Echo(s_agent, "udp_echo", T0, 4, 100, i => new EchoPacket(false, i >= 2, 20_000, 20_000)); + RawChunk chunk = ChunkFactory.Echo(s_agent, "quic_ping", T0, 4, 100, i => new EchoPacket(false, i >= 2, 20_000, 20_000)); Assert.Equal(2, ((RunLengthAcc)EvaluateOneBucket(chunk, "loss.run").Items["loss.run"]).Max); } @@ -114,7 +114,7 @@ public sealed class OpTests { bool[] upLost = [false, true, true, false, false, false, true, true]; bool[] downLost = [false, false, true, true, false, true, false, true]; - RawChunk chunk = ChunkFactory.Echo(s_agent, "udp_echo", T0, upLost.Length, 100, + RawChunk chunk = ChunkFactory.Echo(s_agent, "quic_ping", T0, upLost.Length, 100, i => new EchoPacket(upLost[i], downLost[i], 20_000, 20_000)); MetricSet set = EvaluateOneBucket(chunk, "loss.run.up", "loss.run.down"); @@ -135,7 +135,7 @@ public sealed class OpTests public void Reorder_CountsLateArrivals_AndIgnoresDuplicates() { // 包 3 的回复比包 4 晚到 - RawChunk chunk = ChunkFactory.Echo(s_agent, "udp_echo", T0, 6, 100, i => new EchoPacket(false, false, 20_000, i == 3 ? 150_000 : 20_000)); + RawChunk chunk = ChunkFactory.Echo(s_agent, "quic_ping", T0, 6, 100, i => new EchoPacket(false, false, 20_000, i == 3 ? 150_000 : 20_000)); MetricSet set = EvaluateOneBucket(chunk, "reorder", "dup"); Assert.Equal(1UL, ((CounterAcc)set.Items["reorder"]).Value); Assert.False(set.Items.ContainsKey("dup")); @@ -146,7 +146,7 @@ public sealed class OpTests { // 第 40 个包之后的 10 个包被队头阻塞，全部在 2540 ms 之后几乎同时到达 RawChunk chunk = ChunkFactory.Echo( - s_agent, "tcp_echo", T0, 100, 50, + s_agent, "tcp_ping", T0, 100, 50, i => new EchoPacket(false, false, 20_000, i is >= 40 and < 50 ? ((2540 + (i - 40) - (50 * i) - 20) * 1000) - 50 : 20_000)); MetricSet set = EvaluateOneBucket(chunk, "tcp.stall", "tcp.stall_max"); Assert.Equal(1UL, ((CounterAcc)set.Items["tcp.stall"]).Value); @@ -156,7 +156,7 @@ public sealed class OpTests [Fact] public void Stall_SteadyArrivals_AreNotStalls() { - RawChunk chunk = ChunkFactory.Echo(s_agent, "tcp_echo", T0, 100, 50, _ => new EchoPacket(false, false, 20_000, 20_000)); + RawChunk chunk = ChunkFactory.Echo(s_agent, "tcp_ping", T0, 100, 50, _ => new EchoPacket(false, false, 20_000, 20_000)); MetricSet set = EvaluateOneBucket(chunk, "tcp.stall"); Assert.False(set.Items.ContainsKey("tcp.stall")); } @@ -166,7 +166,7 @@ public sealed class OpTests { // 基线 50 ms（单程 25 ms），阈值 = max(90, 50 + max(15, 17.5)) = 90 ms；第 10、20、30 个回显的 RTT 是 200 ms RawChunk chunk = ChunkFactory.Echo( - s_agent, "tcp_echo", T0, 100, 50, + s_agent, "tcp_ping", T0, 100, 50, i => new EchoPacket(false, false, 25_000, i % 10 == 0 && i > 0 && i < 40 ? 175_000 : 25_000, ProcessUs: 0)); var slow = (RatioAcc)EvaluateOneBucket(chunk, "tcp.slow").Items["tcp.slow"]; Assert.Equal(100UL, slow.Denominator); @@ -176,10 +176,10 @@ public sealed class OpTests [Fact] public void Slow_ASteadyConnection_HasNoLongTail_AndTooFewSamplesGiveNoObservation() { - RawChunk steady = ChunkFactory.Echo(s_agent, "tcp_echo", T0, 100, 50, _ => new EchoPacket(false, false, 25_000, 25_000)); + RawChunk steady = ChunkFactory.Echo(s_agent, "tcp_ping", T0, 100, 50, _ => new EchoPacket(false, false, 25_000, 25_000)); Assert.Equal(0UL, ((RatioAcc)EvaluateOneBucket(steady, "tcp.slow").Items["tcp.slow"]).Numerator); - RawChunk few = ChunkFactory.Echo(s_agent, "tcp_echo", T0, 10, 50, i => new EchoPacket(false, false, 25_000, i == 5 ? 500_000 : 25_000)); + RawChunk few = ChunkFactory.Echo(s_agent, "tcp_ping", T0, 10, 50, i => new EchoPacket(false, false, 25_000, i == 5 ? 500_000 : 25_000)); Assert.False(EvaluateOneBucket(few, "tcp.slow").Items.ContainsKey("tcp.slow")); } @@ -188,7 +188,7 @@ public sealed class OpTests { // 1000 个回显，其中 5 个 1 s 的长尾：p50 约 50 ms，p99 仍在 50 ms 附近，p99.9 落在长尾上 RawChunk chunk = ChunkFactory.Echo( - s_agent, "tcp_echo", T0, 1000, 10, + s_agent, "tcp_ping", T0, 1000, 10, i => new EchoPacket(false, false, 25_000, i % 200 == 0 && i > 0 ? 975_000 : 25_000, ProcessUs: 0), durationMs: 20_000); var rtt = (HistogramAcc)EvaluateOneBucket(chunk, "tcp.rtt").Items["tcp.rtt"]; MetricDef def = TcpMetrics.Rtt; @@ -213,7 +213,7 @@ public sealed class OpTests (baseUs + 1_000_000, 0, 40_000u, 2u, 300u), (baseUs + 2_000_000, 0, 40_000u, 2u, 500u), ]); - RawChunk chunk = ChunkFactory.Echo(s_agent, "tcp_echo", T0, 10, 100, _ => new EchoPacket(false, false, 20_000, 20_000), extraStreams: [info]); + RawChunk chunk = ChunkFactory.Echo(s_agent, "tcp_ping", T0, 10, 100, _ => new EchoPacket(false, false, 20_000, 20_000), extraStreams: [info]); var retrans = (RatioAcc)EvaluateOneBucket(chunk, "tcp.retrans").Items["tcp.retrans"]; Assert.Equal(2UL, retrans.Numerator); Assert.Equal(400UL, retrans.Denominator); @@ -224,7 +224,7 @@ public sealed class OpTests { long baseUs = ChunkFactory.DialerUs(0); RawStream info = ChunkFactory.TcpInfo([(baseUs, 0, 1u, 10u, 100u), (baseUs + 1_000_000, 0, 1u, 0u, 50u), (baseUs + 2_000_000, 0, 1u, 1u, 150u)]); - RawChunk chunk = ChunkFactory.Echo(s_agent, "tcp_echo", T0, 10, 100, _ => new EchoPacket(false, false, 20_000, 20_000), extraStreams: [info]); + RawChunk chunk = ChunkFactory.Echo(s_agent, "tcp_ping", T0, 10, 100, _ => new EchoPacket(false, false, 20_000, 20_000), extraStreams: [info]); var retrans = (RatioAcc)EvaluateOneBucket(chunk, "tcp.retrans").Items["tcp.retrans"]; Assert.Equal(1UL, retrans.Numerator); Assert.Equal(100UL, retrans.Denominator); @@ -234,12 +234,12 @@ public sealed class OpTests public void Retrans_InterleavedConnections_DoNotCreateFalseRetransmissions() { long ts = ChunkFactory.DialerUs(0); - var builder = new StreamBuilder(Kinds.TcpEcho.FindStream("tcp_info")!); + var builder = new StreamBuilder(Kinds.TcpPing.FindStream("tcp_info")!); builder.Add(ts, 0, 40_000, 0, 10, 10, 1000, 0, 1); builder.Add(ts + 1_000_000, 0, 40_000, 0, 10, 20, 1100, 0, 2); builder.Add(ts + 2_000_000, 0, 40_000, 0, 10, 11, 1100, 0, 1); builder.Add(ts + 3_000_000, 0, 40_000, 0, 10, 21, 1200, 0, 2); - RawChunk chunk = ChunkFactory.Echo(s_agent, "tcp_echo", T0, 10, 100, + RawChunk chunk = ChunkFactory.Echo(s_agent, "tcp_ping", T0, 10, 100, _ => new EchoPacket(false, false, 20_000, 20_000), extraStreams: [builder.Build()]); var retrans = (RatioAcc)EvaluateOneBucket(chunk, "tcp.retrans").Items["tcp.retrans"]; Assert.Equal(2UL, retrans.Numerator); @@ -256,7 +256,7 @@ public sealed class OpTests (ts + 1_000_000, 0, 40_000u, 1u, 200u), (ts + 3_000_000, 0, 40_000u, 3u, 400u), ]); - RawChunk chunk = ChunkFactory.Echo(s_agent, "tcp_echo", T0, 10, 100, + RawChunk chunk = ChunkFactory.Echo(s_agent, "tcp_ping", T0, 10, 100, _ => new EchoPacket(false, false, 20_000, 20_000), extraStreams: [info]); var retrans = (RatioAcc)EvaluateOneBucket(chunk, "tcp.retrans").Items["tcp.retrans"]; Assert.Equal(3UL, retrans.Numerator); @@ -267,12 +267,12 @@ public sealed class OpTests public void Retrans_EqualConnectionIdsOnDifferentSides_AreIndependent() { long ts = ChunkFactory.DialerUs(0); - var builder = new StreamBuilder(Kinds.TcpEcho.FindStream("tcp_info")!); + var builder = new StreamBuilder(Kinds.TcpPing.FindStream("tcp_info")!); builder.Add(ts, 0, 40_000, 0, 10, 10, 1000, 0, 1); builder.Add(ts, 1, 40_000, 0, 10, 100, 2000, 0, 1); builder.Add(ts + 1_000_000, 0, 40_000, 0, 10, 11, 1100, 0, 1); builder.Add(ts + 1_000_000, 1, 40_000, 0, 10, 102, 2200, 0, 1); - RawChunk chunk = ChunkFactory.Echo(s_agent, "tcp_echo", T0, 10, 100, + RawChunk chunk = ChunkFactory.Echo(s_agent, "tcp_ping", T0, 10, 100, _ => new EchoPacket(false, false, 20_000, 20_000), extraStreams: [builder.Build()]); var retrans = (RatioAcc)EvaluateOneBucket(chunk, "tcp.retrans").Items["tcp.retrans"]; Assert.Equal(3UL, retrans.Numerator); @@ -283,12 +283,12 @@ public sealed class OpTests public void Retrans_ForwardAndReturnAreSplitBySide() { long ts = ChunkFactory.DialerUs(0); - var builder = new StreamBuilder(Kinds.TcpEcho.FindStream("tcp_info")!); + var builder = new StreamBuilder(Kinds.TcpPing.FindStream("tcp_info")!); builder.Add(ts, 0, 40_000, 0, 10, 10, 1000, 0, 1); builder.Add(ts, 1, 40_000, 0, 10, 100, 2000, 0, 1); builder.Add(ts + 1_000_000, 0, 40_000, 0, 10, 11, 1100, 0, 1); builder.Add(ts + 1_000_000, 1, 40_000, 0, 10, 102, 2200, 0, 1); - RawChunk chunk = ChunkFactory.Echo(s_agent, "tcp_echo", T0, 10, 100, + RawChunk chunk = ChunkFactory.Echo(s_agent, "tcp_ping", T0, 10, 100, _ => new EchoPacket(false, false, 20_000, 20_000), extraStreams: [builder.Build()]); var forward = (RatioAcc)EvaluateOneBucket(chunk, "tcp.retrans.up").Items["tcp.retrans.up"]; var back = (RatioAcc)EvaluateOneBucket(chunk, "tcp.retrans.down").Items["tcp.retrans.down"]; @@ -322,9 +322,9 @@ public sealed class OpTests // 10 秒的块，按 1 秒分桶：每个桶 10 个包 RawChunk chunk = ChunkFactory.UdpEchoClean(s_agent, T0); var buckets = new SortedDictionary(); - WindowEvaluator.Default.Evaluate(chunk, 1000, buckets, new HashSet { "n.udp_echo" }); + WindowEvaluator.Default.Evaluate(chunk, 1000, buckets, new HashSet { "n.quic_ping" }); Assert.Equal(10, buckets.Count); - Assert.All(buckets.Values, set => Assert.Equal(10UL, ((CounterAcc)set.Items["n.udp_echo"]).Value)); + Assert.All(buckets.Values, set => Assert.Equal(10UL, ((CounterAcc)set.Items["n.quic_ping"]).Value)); Assert.Equal(T0, buckets.Keys.First()); } diff --git a/backend/tests/MoShiro.Backend.Tests/AgentDataPlaneTests.cs b/backend/tests/MoShiro.Backend.Tests/AgentDataPlaneTests.cs index c2b8959..2a702d1 100644 --- a/backend/tests/MoShiro.Backend.Tests/AgentDataPlaneTests.cs +++ b/backend/tests/MoShiro.Backend.Tests/AgentDataPlaneTests.cs @@ -22,8 +22,8 @@ public sealed class AgentDataPlaneTests : IClassFixture AgentHarness agent = await AgentHarness.StartAsync(_backend); await agent.OverrideLineTasksAsync(rateHz: 30); DialerSimulator dialer = await agent.ConnectDialerAsync(); - LineTaskSet set = await dialer.WaitForTasksAsync(s => s.Tasks.Any(t => t.Kind == "udp_echo" && t.Port == agent.App.QuicPorts[0])); - return (agent, dialer, set.Tasks.Single(t => t.Kind == "udp_echo")); + LineTaskSet set = await dialer.WaitForTasksAsync(s => s.Tasks.Any(t => t.Kind == "quic_ping" && t.Port == agent.App.QuicPorts[0])); + return (agent, dialer, set.Tasks.Single(t => t.Kind == "quic_ping")); } [Fact] diff --git a/backend/tests/MoShiro.Backend.Tests/AgentEndToEndTests.cs b/backend/tests/MoShiro.Backend.Tests/AgentEndToEndTests.cs index d489a8d..03ab5aa 100644 --- a/backend/tests/MoShiro.Backend.Tests/AgentEndToEndTests.cs +++ b/backend/tests/MoShiro.Backend.Tests/AgentEndToEndTests.cs @@ -23,8 +23,8 @@ public sealed class AgentEndToEndTests : IClassFixture AgentHarness agent = await AgentHarness.StartAsync(_backend); await agent.OverrideLineTasksAsync(rateHz: 30); DialerSimulator dialer = await agent.ConnectDialerAsync(); - LineTaskSet set = await dialer.WaitForTasksAsync(s => s.Tasks.Any(t => t.Kind == "udp_echo" && t.Port == agent.App.QuicPorts[0])); - return (agent, dialer, set.Tasks.Single(t => t.Kind == "udp_echo")); + LineTaskSet set = await dialer.WaitForTasksAsync(s => s.Tasks.Any(t => t.Kind == "quic_ping" && t.Port == agent.App.QuicPorts[0])); + return (agent, dialer, set.Tasks.Single(t => t.Kind == "quic_ping")); } private async Task AgentViewAsync(AgentHarness agent) => await _backend.GetJsonAsync(_backend.Admin, "/api/admin/agents/" + agent.Id); @@ -61,7 +61,7 @@ public sealed class AgentEndToEndTests : IClassFixture }, because: (await _backend.GetJsonAsync(_backend.Admin, $"/api/admin/events?agent_id={agent.Id}")).ToJsonString() + $" fragments={agent.App.Counters.Fragments} discarded={agent.App.Counters.FragmentsDiscarded} outbox={agent.App.Link.Outbox.PendingCount}"); JsonNode chunk = stored["items"]![0]!; Assert.True(chunk["kind"] is not null, stored.ToJsonString()); - Assert.Equal("udp_echo", chunk["kind"]!.GetValue()); + Assert.Equal("quic_ping", chunk["kind"]!.GetValue()); Assert.True(agent.App.Counters.Fragments >= 1); Assert.True(agent.App.Counters.PacketsIn >= 40); } @@ -76,11 +76,11 @@ public sealed class AgentEndToEndTests : IClassFixture await using (agent) await using (dialer) { - LineTask tcp = (await dialer.WaitForTasksAsync(s => s.Tasks.Any(t => t.Kind == "tcp_echo" && t.Port == agent.App.TcpPorts[0]))).Tasks.Single(t => t.Kind == "tcp_echo"); + LineTask tcp = (await dialer.WaitForTasksAsync(s => s.Tasks.Any(t => t.Kind == "tcp_ping" && t.Port == agent.App.TcpPorts[0]))).Tasks.Single(t => t.Kind == "tcp_ping"); (int sent, int received) = await dialer.RunTcpEchoAsync(tcp, agent.App.TcpPorts[0], 15, TimeSpan.FromMilliseconds(60), tls); Assert.Equal(15, sent); Assert.Equal(15, received); - await BackendFixture.EventuallyAsync(async () => (await _backend.GetJsonAsync(_backend.Admin, $"/api/admin/raw/chunks?agent_id={agent.Id}"))["items"]!.AsArray().Any(c => c!["kind"]!.GetValue() == "tcp_echo")); + await BackendFixture.EventuallyAsync(async () => (await _backend.GetJsonAsync(_backend.Admin, $"/api/admin/raw/chunks?agent_id={agent.Id}"))["items"]!.AsArray().Any(c => c!["kind"]!.GetValue() == "tcp_ping")); } } @@ -98,8 +98,8 @@ public sealed class AgentEndToEndTests : IClassFixture // The dialer process restarts: a new session with its own streams; sequence numbers start from 0 again await using DialerSimulator second = first.Restarted(); await second.ConnectAsync("127.0.0.1", agent.App.ControlPort, agent.TlsPublicKey, agent.App.QuicPorts[0], agent.NodeId); - LineTaskSet set = await second.WaitForTasksAsync(s => s.Tasks.Any(t => t.Kind == "udp_echo" && t.Port == agent.App.QuicPorts[0])); - LineTask udp2 = set.Tasks.Single(t => t.Kind == "udp_echo"); + LineTaskSet set = await second.WaitForTasksAsync(s => s.Tasks.Any(t => t.Kind == "quic_ping" && t.Port == agent.App.QuicPorts[0])); + LineTask udp2 = set.Tasks.Single(t => t.Kind == "quic_ping"); Assert.NotEqual(udp.StreamId, udp2.StreamId); // The previous session's streams are released with it ... @@ -135,9 +135,9 @@ public sealed class AgentEndToEndTests : IClassFixture Assert.Equal(agent.Id, dialerView["outgoing"]![0]!["listener_id"]!.GetValue()); // Both get their own task set (own streams) - LineTaskSet setA = await a.WaitForTasksAsync(s => s.Tasks.Any(t => t.Kind == "udp_echo")); - LineTaskSet setB = await b.WaitForTasksAsync(s => s.Tasks.Any(t => t.Kind == "udp_echo")); - Assert.NotEqual(setA.Tasks.Single(t => t.Kind == "udp_echo").StreamId, setB.Tasks.Single(t => t.Kind == "udp_echo").StreamId); + LineTaskSet setA = await a.WaitForTasksAsync(s => s.Tasks.Any(t => t.Kind == "quic_ping")); + LineTaskSet setB = await b.WaitForTasksAsync(s => s.Tasks.Any(t => t.Kind == "quic_ping")); + Assert.NotEqual(setA.Tasks.Single(t => t.Kind == "quic_ping").StreamId, setB.Tasks.Single(t => t.Kind == "quic_ping").StreamId); // Drop one: the other stays using HttpResponseMessage dropped = await _backend.Admin.PostAsync($"/api/admin/agents/{agent.Id}/dialers/{a.DialerId}/disconnect", null); @@ -172,7 +172,7 @@ public sealed class AgentEndToEndTests : IClassFixture await using AgentHarness agent = await AgentHarness.StartAsync(_backend, config => config.Limits.MaxTotalMbps = 20); await agent.OverrideLineTasksAsync(); - var phase = new TestPhase { Kind = "udp_train", DurationSeconds = 1, Params = { ["port"] = System.Text.Json.JsonSerializer.SerializeToElement(agent.App.QuicPorts[0]) } }; + var phase = new TestPhase { Kind = "quic_stream", DurationSeconds = 1, Params = { ["port"] = System.Text.Json.JsonSerializer.SerializeToElement(agent.App.QuicPorts[0]) } }; var run = new TestRunParams { TestId = new TestId("tst_x"), DialerId = new NodeId("agt_nobody"), Plan = "x", Phases = [phase] }; Assert.Equal("dialer_offline", agent.App.Tests.Start(run).Error.Code.Value); diff --git a/backend/tests/MoShiro.Backend.Tests/AgentHarness.cs b/backend/tests/MoShiro.Backend.Tests/AgentHarness.cs index 078d944..7748506 100644 --- a/backend/tests/MoShiro.Backend.Tests/AgentHarness.cs +++ b/backend/tests/MoShiro.Backend.Tests/AgentHarness.cs @@ -105,19 +105,19 @@ public sealed class AgentHarness : IAsyncDisposable public async Task OverrideLineTasksAsync(int? rateHz = null) { JsonNode tasks = await Backend.GetJsonAsync(Backend.Admin, "/api/admin/tasks"); - // 先处理 connect（关掉），再改 udp_echo / tcp_echo 的速率：否则中途的带宽预算会多出 connect 那一份 - foreach (JsonNode? task in tasks["items"]!.AsArray().OrderBy(t => t!["kind"]!.GetValue() == "connect" ? 0 : 1)) + // 先处理 connect（关掉），再改 quic_ping / tcp_ping 的速率：否则中途的带宽预算会多出 connect 那一份 + foreach (JsonNode? task in tasks["items"]!.AsArray().OrderBy(t => t!["kind"]!.GetValue() == "tcp_connect" ? 0 : 1)) { - if (task!["scope"]!.GetValue() != "line" || task["agent_id"] is not null || task["kind"]!.GetValue() is not ("udp_echo" or "tcp_echo" or "connect")) + if (task!["scope"]!.GetValue() != "line" || task["agent_id"] is not null || task["kind"]!.GetValue() is not ("quic_ping" or "tcp_ping" or "tcp_connect")) { continue; } string kind = task["kind"]!.GetValue(); - var parameters = new JsonObject { ["port"] = kind == "udp_echo" ? App.QuicPorts[0] : App.TcpPorts[0] }; + var parameters = new JsonObject { ["port"] = kind == "quic_ping" ? App.QuicPorts[0] : App.TcpPorts[0] }; // 默认的 connect 模板在这些用例里不需要（也占预算）：对这个 Agent 关掉它 - bool enabled = kind != "connect"; - if (rateHz is { } hz && kind != "connect") + bool enabled = kind != "tcp_connect"; + if (rateHz is { } hz && kind != "tcp_connect") { parameters["adaptive"] = false; // 固定频率：用例按这个速率发包，限速也按它算 parameters["rate_hz"] = hz; diff --git a/backend/tests/MoShiro.Backend.Tests/AgentMeshTests.cs b/backend/tests/MoShiro.Backend.Tests/AgentMeshTests.cs index 87a9beb..b38727a 100644 --- a/backend/tests/MoShiro.Backend.Tests/AgentMeshTests.cs +++ b/backend/tests/MoShiro.Backend.Tests/AgentMeshTests.cs @@ -41,7 +41,7 @@ public sealed class AgentMeshTests : IClassFixture await BackendFixture.EventuallyAsync(async () => { JsonNode chunks = await _backend.GetJsonAsync(_backend.Admin, $"/api/admin/raw/chunks?agent_id={listener.Id}"); - return chunks["items"]!.AsArray().Any(c => c!["kind"]!.GetValue() == "udp_echo" && c["series"]!.GetValue() == "line:" + dialer.Id); + return chunks["items"]!.AsArray().Any(c => c!["kind"]!.GetValue() == "quic_ping" && c["series"]!.GetValue() == "line:" + dialer.Id); }, TimeSpan.FromSeconds(40), "the line should persist measured packet data"); } } diff --git a/backend/tests/MoShiro.Backend.Tests/AgentTaskSetEndToEndTests.cs b/backend/tests/MoShiro.Backend.Tests/AgentTaskSetEndToEndTests.cs index 01c743e..c0c5ba6 100644 --- a/backend/tests/MoShiro.Backend.Tests/AgentTaskSetEndToEndTests.cs +++ b/backend/tests/MoShiro.Backend.Tests/AgentTaskSetEndToEndTests.cs @@ -22,15 +22,15 @@ public sealed class AgentTaskSetEndToEndTests : IClassFixture await using AgentHarness agent = await AgentHarness.StartAsync(_backend); await agent.OverrideLineTasksAsync(rateHz: 30); await using DialerSimulator dialer = await agent.ConnectDialerAsync(); - LineTask udp = (await dialer.WaitForTasksAsync(s => s.Tasks.Any(t => t.Kind == "udp_echo" && t.Port == agent.App.QuicPorts[0]))) - .Tasks.Single(t => t.Kind == "udp_echo"); + LineTask udp = (await dialer.WaitForTasksAsync(s => s.Tasks.Any(t => t.Kind == "quic_ping" && t.Port == agent.App.QuicPorts[0]))) + .Tasks.Single(t => t.Kind == "quic_ping"); JsonNode tasks = await _backend.GetJsonAsync(_backend.Admin, "/api/admin/tasks"); - JsonNode mine = tasks["items"]!.AsArray().Single(t => t!["agent_id"]?.GetValue() == agent.Id && t["kind"]!.GetValue() == "udp_echo")!; + JsonNode mine = tasks["items"]!.AsArray().Single(t => t!["agent_id"]?.GetValue() == agent.Id && t["kind"]!.GetValue() == "quic_ping")!; using HttpResponseMessage disabled = await _backend.Admin.PatchAsync("/api/admin/tasks/" + mine["id"]!.GetValue(), BackendFixture.Json(new JsonObject { ["enabled"] = false }, "application/merge-patch+json")); disabled.EnsureSuccessStatusCode(); - LineTaskSet next = await dialer.WaitForTasksAsync(s => s.Tasks.All(t => t.Kind != "udp_echo") && s.Version > 0); + LineTaskSet next = await dialer.WaitForTasksAsync(s => s.Tasks.All(t => t.Kind != "quic_ping") && s.Version > 0); Assert.DoesNotContain(next.Tasks, t => t.Id == udp.Id); } } diff --git a/backend/tests/MoShiro.Backend.Tests/AotSmokeTests.cs b/backend/tests/MoShiro.Backend.Tests/AotSmokeTests.cs index b091597..9ce1f8b 100644 --- a/backend/tests/MoShiro.Backend.Tests/AotSmokeTests.cs +++ b/backend/tests/MoShiro.Backend.Tests/AotSmokeTests.cs @@ -58,28 +58,28 @@ public sealed class AotSmokeTests string lineId = $"{agent.NodeId.Value}~{dialerId}"; await agent.ConnectAsync(baseAddress); TaskSet tasks = await agent.WaitForTaskSetAsync(s => s.Tasks.Count >= 2); - Assert.Contains(tasks.Tasks, t => t.Kind == "udp_echo"); + Assert.Contains(tasks.Tasks, t => t.Kind == "quic_ping"); // 原始块：二进制通道 → 段文件与索引 → ACK long minute = 60_000; long baseMs = ((TimeProvider.System.GetUtcNow().ToUnixTimeMilliseconds() / minute) - 3) * minute; for (int k = 0; k < 6; k++) { - Assert.True(await agent.SendChunkAsync(ChunkFactory.Echo(agent.NodeId, "udp_echo", baseMs + (k * 10_000L), 100, 100, i => new EchoPacket(false, i % 10 == 0, 20_000, 20_000)))); + Assert.True(await agent.SendChunkAsync(ChunkFactory.Echo(agent.NodeId, "quic_ping", baseMs + (k * 10_000L), 100, 100, i => new EchoPacket(false, i % 10 == 0, 20_000, 20_000)))); } // 查询：热层的精确汇总，随后是持久层的分钟窗口 string range = $"from={Iso(baseMs)}&to={Iso(baseMs + minute)}&metrics=loss.total,rtt.p95"; JsonNode hot = await GetAsync(http, $"/api/public/lines/{lineId}/series?{range}&step=1s"); Assert.Equal("hot", hot["source"]!.GetValue()); - Assert.Equal(600, hot["summary"]!["samples"]!["udp_echo"]!.GetValue()); + Assert.Equal(600, hot["summary"]!["samples"]!["quic_ping"]!.GetValue()); Assert.Equal(10.0, hot["summary"]!["metrics"]!["loss.total"]!.GetValue(), 6); long deadline = TimeProvider.System.GetTimestamp(); JsonNode persisted; do { persisted = await GetAsync(http, $"/api/public/lines/{lineId}/series?{range}&step=1m"); - if (persisted["summary"]!["samples"]!["udp_echo"]?.GetValue() == 600) + if (persisted["summary"]!["samples"]!["quic_ping"]?.GetValue() == 600) { break; } @@ -88,7 +88,7 @@ public sealed class AotSmokeTests } while (TimeProvider.System.GetElapsedTime(deadline) < TimeSpan.FromSeconds(15)); Assert.Equal("1m", persisted["source"]!.GetValue()); - Assert.Equal(600, persisted["summary"]!["samples"]!["udp_echo"]!.GetValue()); + Assert.Equal(600, persisted["summary"]!["samples"]!["quic_ping"]!.GetValue()); // 原始块可以按 ID 下载并解码 JsonNode chunks = await GetAsync(http, $"/api/admin/raw/chunks?agent_id={agent.NodeId.Value}&limit=1"); diff --git a/backend/tests/MoShiro.Backend.Tests/ApiBasicsTests.cs b/backend/tests/MoShiro.Backend.Tests/ApiBasicsTests.cs index 20b4f0a..83f4018 100644 --- a/backend/tests/MoShiro.Backend.Tests/ApiBasicsTests.cs +++ b/backend/tests/MoShiro.Backend.Tests/ApiBasicsTests.cs @@ -194,7 +194,7 @@ public sealed class ApiBasicsTests : IClassFixture Assert.Equal(HttpStatusCode.NotModified, notModified.StatusCode); JsonNode kinds = await _backend.GetJsonAsync(_backend.Anonymous, "/api/public/catalog/kinds"); - JsonNode udp = kinds["items"]!.AsArray().Single(k => k!["id"]!.GetValue() == "udp_echo")!; + JsonNode udp = kinds["items"]!.AsArray().Single(k => k!["id"]!.GetValue() == "quic_ping")!; JsonNode rate = udp["params"]!.AsArray().Single(p => p!["name"]!.GetValue() == "rate_hz")!; Assert.Equal(1, rate["min"]!.GetValue()); Assert.Equal(200, rate["max"]!.GetValue()); diff --git a/backend/tests/MoShiro.Backend.Tests/BackendFixture.cs b/backend/tests/MoShiro.Backend.Tests/BackendFixture.cs index 45943b9..5ede417 100644 --- a/backend/tests/MoShiro.Backend.Tests/BackendFixture.cs +++ b/backend/tests/MoShiro.Backend.Tests/BackendFixture.cs @@ -11,7 +11,7 @@ using MoShiro.TestSupport; namespace MoShiro.Backend.Tests; -/// 首次启动种入的默认线路任务模板数：udp_echo、tcp_echo、connect。 +/// 首次启动种入的默认线路任务模板数：quic_ping、tcp_ping、connect。 internal static class Defaults { public const int TaskCount = 3; diff --git a/backend/tests/MoShiro.Backend.Tests/DialerEndToEndTests.cs b/backend/tests/MoShiro.Backend.Tests/DialerEndToEndTests.cs index 3abadf6..b676d34 100644 --- a/backend/tests/MoShiro.Backend.Tests/DialerEndToEndTests.cs +++ b/backend/tests/MoShiro.Backend.Tests/DialerEndToEndTests.cs @@ -59,7 +59,7 @@ public sealed class DialerEndToEndTests : IClassFixture TimeSpan.FromSeconds(90)); Assert.True(test["status"]!.GetValue() == "succeeded", test.ToJsonString()); - await BackendFixture.EventuallyAsync(async () => await ChunkCountAsync(agent, "udp_train") >= 5, because: "the 100 Mbit/s phase (5 s of 1000 B packets, over 50000 records) is split into two fragments"); + await BackendFixture.EventuallyAsync(async () => await ChunkCountAsync(agent, "quic_stream") >= 5, because: "the 100 Mbit/s phase (5 s of 1000 B packets, over 50000 records) is split into two fragments"); Assert.Equal(0, agent.App.Counters.FragmentsDiscarded); } @@ -84,7 +84,7 @@ public sealed class DialerEndToEndTests : IClassFixture }, TimeSpan.FromSeconds(150)); Assert.True(test["status"]!.GetValue() == "succeeded", test.ToJsonString()); - await BackendFixture.EventuallyAsync(async () => await ChunkCountAsync(agent, "tcp_bulk") >= 1, because: "the bulk phase stores its fragments"); + await BackendFixture.EventuallyAsync(async () => await ChunkCountAsync(agent, "tcp_transfer") >= 1, because: "the bulk phase stores its fragments"); Assert.Equal(0, agent.App.Counters.FragmentsDiscarded); } } diff --git a/backend/tests/MoShiro.Backend.Tests/EnrollmentAndTaskSetTests.cs b/backend/tests/MoShiro.Backend.Tests/EnrollmentAndTaskSetTests.cs index f4a50ad..e780cc8 100644 --- a/backend/tests/MoShiro.Backend.Tests/EnrollmentAndTaskSetTests.cs +++ b/backend/tests/MoShiro.Backend.Tests/EnrollmentAndTaskSetTests.cs @@ -24,15 +24,15 @@ public sealed class EnrollmentAndTaskSetTests : IClassFixture Assert.Equal("active", agent.EnrollmentStatus); Assert.Equal("active", agent.AuthStatus); TaskSet set = await agent.WaitForTaskSetAsync(s => s.Tasks.Count == Defaults.TaskCount); - Assert.Equal(["connect", "tcp_echo", "udp_echo"], set.Tasks.Select(t => t.Kind).Order().ToArray()); + Assert.Equal(["quic_ping", "tcp_connect", "tcp_ping"], set.Tasks.Select(t => t.Kind).Order().ToArray()); Assert.All(set.Tasks, t => { Assert.Equal("line", t.Scope); Assert.Equal(TaskSchedule.Continuous, t.Schedule); Assert.NotEmpty(t.Params); // 参数已经补全了默认值 }); - Assert.Equal(10, set.Tasks.Single(t => t.Kind == "udp_echo").Params["rate_hz"].GetInt64()); - Assert.Equal(7444, set.Tasks.Single(t => t.Kind == "tcp_echo").Params["port"].GetInt64()); + Assert.Equal(10, set.Tasks.Single(t => t.Kind == "quic_ping").Params["rate_hz"].GetInt64()); + Assert.Equal(7444, set.Tasks.Single(t => t.Kind == "tcp_ping").Params["port"].GetInt64()); Assert.True(set.Version >= 1); JsonNode view = await AgentAsync(agent.NodeId.Value); @@ -46,9 +46,9 @@ public sealed class EnrollmentAndTaskSetTests : IClassFixture [Fact] public async Task TaskSet_OnlyContainsKindsTheAgentSupports() { - await using AgentSimulator agent = await _backend.ConnectAgentAsync(kinds: ["udp_echo"]); + await using AgentSimulator agent = await _backend.ConnectAgentAsync(kinds: ["quic_ping"]); TaskSet set = await agent.WaitForTaskSetAsync(s => s.Tasks.Count >= 1); - Assert.Equal(["udp_echo"], set.Tasks.Select(t => t.Kind).ToArray()); + Assert.Equal(["quic_ping"], set.Tasks.Select(t => t.Kind).ToArray()); } [Fact] @@ -265,12 +265,12 @@ public sealed class EnrollmentAndTaskSetTests : IClassFixture { await using AgentSimulator agent = await _backend.ConnectAgentAsync(); TaskSet initial = await agent.WaitForTaskSetAsync(s => s.Tasks.Count == Defaults.TaskCount); - TaskSpec udp = initial.Tasks.Single(t => t.Kind == "udp_echo"); + TaskSpec udp = initial.Tasks.Single(t => t.Kind == "quic_ping"); using HttpResponseMessage created = await _backend.Admin.PostAsync("/api/admin/tasks", BackendFixture.Json(new JsonObject { ["scope"] = "line", - ["kind"] = "udp_echo", + ["kind"] = "quic_ping", ["agent_id"] = agent.NodeId.Value, ["base_task_id"] = udp.Id.Value, ["params"] = new JsonObject { ["rate_hz"] = 20 }, @@ -278,15 +278,15 @@ public sealed class EnrollmentAndTaskSetTests : IClassFixture Assert.Equal(HttpStatusCode.Created, created.StatusCode); Assert.StartsWith("/api/admin/tasks/", created.Headers.Location!.ToString()); - TaskSet updated = await agent.WaitForTaskSetAsync(s => s.Version > initial.Version && s.Tasks.Any(t => t.Kind == "udp_echo" && t.Params["rate_hz"].GetInt64() == 20)); + TaskSet updated = await agent.WaitForTaskSetAsync(s => s.Version > initial.Version && s.Tasks.Any(t => t.Kind == "quic_ping" && t.Params["rate_hz"].GetInt64() == 20)); Assert.Equal(Defaults.TaskCount, updated.Tasks.Count); - Assert.Equal(7443, updated.Tasks.Single(t => t.Kind == "udp_echo").Params["port"].GetInt64()); + Assert.Equal(7443, updated.Tasks.Single(t => t.Kind == "quic_ping").Params["port"].GetInt64()); // 停用覆盖 → 该 Agent 不再运行这条任务（模板对其他 Agent 不受影响） string overrideId = JsonNode.Parse(await created.Content.ReadAsStringAsync())!["id"]!.GetValue(); using HttpResponseMessage disable = await _backend.Admin.PatchAsync("/api/admin/tasks/" + overrideId, BackendFixture.Json(new JsonObject { ["enabled"] = false }, "application/merge-patch+json")); Assert.Equal(HttpStatusCode.OK, disable.StatusCode); - TaskSet disabled = await agent.WaitForTaskSetAsync(s => s.Version > updated.Version && s.Tasks.All(t => t.Kind != "udp_echo")); + TaskSet disabled = await agent.WaitForTaskSetAsync(s => s.Version > updated.Version && s.Tasks.All(t => t.Kind != "quic_ping")); Assert.Equal(Defaults.TaskCount - 1, disabled.Tasks.Count); } @@ -295,7 +295,7 @@ public sealed class EnrollmentAndTaskSetTests : IClassFixture { using HttpResponseMessage badParams = await _backend.Admin.PostAsync("/api/admin/tasks", BackendFixture.Json(new JsonObject { - ["scope"] = "line", ["kind"] = "udp_echo", ["params"] = new JsonObject { ["rate_hz"] = 5000, ["bogus"] = 1 }, + ["scope"] = "line", ["kind"] = "quic_ping", ["params"] = new JsonObject { ["rate_hz"] = 5000, ["bogus"] = 1 }, })); Assert.Equal(HttpStatusCode.BadRequest, badParams.StatusCode); JsonNode problem = JsonNode.Parse(await badParams.Content.ReadAsStringAsync())!; @@ -308,33 +308,107 @@ public sealed class EnrollmentAndTaskSetTests : IClassFixture using HttpResponseMessage overBudget = await _backend.Admin.PostAsync("/api/admin/tasks", BackendFixture.Json(new JsonObject { - ["scope"] = "line", ["kind"] = "udp_echo", ["params"] = new JsonObject { ["rate_hz"] = 200, ["size"] = 1400 }, + ["scope"] = "line", ["kind"] = "quic_ping", ["params"] = new JsonObject { ["rate_hz"] = 200, ["size"] = 1400 }, })); Assert.Equal(HttpStatusCode.BadRequest, overBudget.StatusCode); Assert.Equal("budget_exceeded", JsonNode.Parse(await overBudget.Content.ReadAsStringAsync())!["code"]!.GetValue()); - using HttpResponseMessage wrongScope = await _backend.Admin.PostAsync("/api/admin/tasks", BackendFixture.Json(new JsonObject { ["scope"] = "line", ["kind"] = "icmp_echo" })); + using HttpResponseMessage wrongScope = await _backend.Admin.PostAsync("/api/admin/tasks", BackendFixture.Json(new JsonObject { ["scope"] = "line", ["kind"] = "icmp_ping" })); Assert.Equal(HttpStatusCode.BadRequest, wrongScope.StatusCode); - // 目标任务：只接受 IP（不做 DNS 解析），需要指定 Agent + // 单个 Agent 的任务：目标的格式按 Kind 检查（Backend 不解析域名），需要指定 Agent await using AgentSimulator agent = await _backend.ConnectAgentAsync(); - using HttpResponseMessage dns = await _backend.Admin.PostAsync("/api/admin/tasks", BackendFixture.Json(new JsonObject + foreach ((string kind, string badTarget) in new[] { ("icmp_ping", "https://example.com/"), ("icmp_ping", "not a host"), ("http_request", "example.com") }) { - ["scope"] = "target", ["kind"] = "icmp_echo", ["agent_id"] = agent.NodeId.Value, ["target"] = "example.com", + using HttpResponseMessage malformed = await _backend.Admin.PostAsync("/api/admin/tasks", BackendFixture.Json(new JsonObject + { + ["scope"] = "target", ["kind"] = kind, ["agent_id"] = agent.NodeId.Value, ["target"] = badTarget, + })); + Assert.Equal(HttpStatusCode.BadRequest, malformed.StatusCode); + } + + using HttpResponseMessage page = await _backend.Admin.PostAsync("/api/admin/tasks", BackendFixture.Json(new JsonObject + { + ["scope"] = "target", ["kind"] = "http_request", ["agent_id"] = agent.NodeId.Value, ["target"] = "https://Example.com/health", ["params"] = new JsonObject { ["version"] = "h3" }, })); - Assert.Equal(HttpStatusCode.BadRequest, dns.StatusCode); + Assert.Equal(HttpStatusCode.Created, page.StatusCode); + Assert.Equal("https://example.com/health", JsonNode.Parse(await page.Content.ReadAsStringAsync())!["target"]!.GetValue()); using HttpResponseMessage target = await _backend.Admin.PostAsync("/api/admin/tasks", BackendFixture.Json(new JsonObject { - ["scope"] = "target", ["kind"] = "icmp_echo", ["agent_id"] = agent.NodeId.Value, ["target"] = "1.1.1.1", + ["scope"] = "target", ["kind"] = "icmp_ping", ["agent_id"] = agent.NodeId.Value, ["target"] = "1.1.1.1", })); Assert.Equal(HttpStatusCode.Created, target.StatusCode); - TaskSet set = await agent.WaitForTaskSetAsync(s => s.Tasks.Any(t => t.Scope == "target")); - Assert.Equal("1.1.1.1", set.Tasks.Single(t => t.Scope == "target").Target); + TaskSet set = await agent.WaitForTaskSetAsync(s => s.Tasks.Any(t => t.Scope == "target" && t.Kind == "icmp_ping")); + Assert.Equal("1.1.1.1", set.Tasks.Single(t => t.Scope == "target" && t.Kind == "icmp_ping").Target); - using HttpResponseMessage onceSchedule = await _backend.Admin.PostAsync("/api/admin/tasks", BackendFixture.Json(new JsonObject { ["scope"] = "line", ["kind"] = "udp_echo", ["schedule"] = "every:30" })); + using HttpResponseMessage onceSchedule = await _backend.Admin.PostAsync("/api/admin/tasks", BackendFixture.Json(new JsonObject { ["scope"] = "line", ["kind"] = "quic_ping", ["schedule"] = "every:30" })); Assert.Equal(HttpStatusCode.BadRequest, onceSchedule.StatusCode); } + [Fact] + public async Task TargetTemplate_AppliesToEveryAgent_AndAnOverrideCanDisableOrRetargetItForOne() + { + await using AgentSimulator first = await _backend.ConnectAgentAsync(); + await using AgentSimulator second = await _backend.ConnectAgentAsync(); + using HttpResponseMessage created = await _backend.Admin.PostAsync("/api/admin/tasks", BackendFixture.Json(new JsonObject + { + ["scope"] = "target", ["kind"] = "tcp_connect", ["target"] = "Example.com:443", + })); + Assert.Equal(HttpStatusCode.Created, created.StatusCode); + JsonNode template = JsonNode.Parse(await created.Content.ReadAsStringAsync())!; + string templateId = template["id"]!.GetValue(); + Assert.Null(template["agent_id"]); + try + { + foreach (AgentSimulator agent in new[] { first, second }) + { + TaskSet set = await agent.WaitForTaskSetAsync(s => s.Tasks.Any(t => t.Kind == "tcp_connect" && t.Scope == "target")); + Assert.Equal("example.com:443", set.Tasks.Single(t => t.Kind == "tcp_connect" && t.Scope == "target").Target); + } + + // 展示接口列出每个 Agent 实际执行的目标；模拟的 Agent 没有上传数据，所以是「等待数据」 + JsonNode listed = await _backend.GetJsonAsync(_backend.Admin, "/api/public/targets"); + string[] both = [first.NodeId.Value, second.NodeId.Value]; + JsonNode[] mine = [.. listed["items"]!.AsArray().Select(i => i!).Where(i => i["task_id"]!.GetValue() == templateId && both.Contains(i["agent_id"]!.GetValue()))]; + Assert.Equal(both.Order(), mine.Select(i => i["agent_id"]!.GetValue()).Order()); + Assert.All(mine, i => + { + Assert.Equal("example.com:443", i["target"]!.GetValue()); + Assert.Equal("waiting", i["status"]!.GetValue()); + Assert.Null(i["last_sample"]); + }); + + JsonNode one = await _backend.GetJsonAsync(_backend.Admin, $"/api/public/targets?agent_id={first.NodeId.Value}&task_id={templateId}&full=true"); + Assert.Equal(first.NodeId.Value, Assert.Single(one["items"]!.AsArray())!["agent_id"]!.GetValue()); + + // 覆盖：第一个 Agent 换一个目标，第二个 Agent 停用 + using HttpResponseMessage retarget = await _backend.Admin.PostAsync("/api/admin/tasks", BackendFixture.Json(new JsonObject + { + ["scope"] = "target", ["kind"] = "tcp_connect", ["agent_id"] = first.NodeId.Value, ["base_task_id"] = templateId, ["target"] = "192.0.2.9:8443", + })); + Assert.Equal(HttpStatusCode.Created, retarget.StatusCode); + using HttpResponseMessage off = await _backend.Admin.PostAsync("/api/admin/tasks", BackendFixture.Json(new JsonObject + { + ["scope"] = "target", ["kind"] = "tcp_connect", ["agent_id"] = second.NodeId.Value, ["base_task_id"] = templateId, ["enabled"] = false, + })); + Assert.Equal(HttpStatusCode.Created, off.StatusCode); + TaskSet firstSet = await first.WaitForTaskSetAsync(s => s.Tasks.Any(t => t.Target == "192.0.2.9:8443")); + Assert.Single(firstSet.Tasks, t => t.Kind == "tcp_connect" && t.Scope == "target"); + await second.WaitForTaskSetAsync(s => s.Tasks.All(t => t.Scope != "target")); + + // 覆盖只能指向同类的单个 Agent 模板 + using HttpResponseMessage wrongKind = await _backend.Admin.PostAsync("/api/admin/tasks", BackendFixture.Json(new JsonObject + { + ["scope"] = "target", ["kind"] = "icmp_ping", ["agent_id"] = first.NodeId.Value, ["base_task_id"] = templateId, + })); + Assert.Equal(HttpStatusCode.BadRequest, wrongKind.StatusCode); + } + finally + { + using HttpResponseMessage deleted = await _backend.Admin.DeleteAsync("/api/admin/tasks/" + templateId); + } + } + [Fact] public async Task TaskSet_RejectedItemsReportedByTheAgent_AreVisibleToTheAdmin() { @@ -344,7 +418,7 @@ public sealed class EnrollmentAndTaskSetTests : IClassFixture using HttpResponseMessage created = await _backend.Admin.PostAsync("/api/admin/tasks", BackendFixture.Json(new JsonObject { - ["scope"] = "target", ["kind"] = "connect", ["agent_id"] = agent.NodeId.Value, ["target"] = "192.0.2.1:443", + ["scope"] = "target", ["kind"] = "tcp_connect", ["agent_id"] = agent.NodeId.Value, ["target"] = "192.0.2.1:443", })); Assert.Equal(HttpStatusCode.Created, created.StatusCode); await BackendFixture.EventuallyAsync(async () => (await AgentAsync(agent.NodeId.Value))["rejected"]!.AsArray().Count == 1); diff --git a/backend/tests/MoShiro.Backend.Tests/IngestAndQueryTests.cs b/backend/tests/MoShiro.Backend.Tests/IngestAndQueryTests.cs index 153d638..5873148 100644 --- a/backend/tests/MoShiro.Backend.Tests/IngestAndQueryTests.cs +++ b/backend/tests/MoShiro.Backend.Tests/IngestAndQueryTests.cs @@ -29,7 +29,7 @@ public sealed class IngestAndQueryTests : IClassFixture { long start = baseMs + (k * 10_000L); RawChunk udp = ChunkFactory.Echo( - agent.NodeId, "udp_echo", start, 100, 100, + agent.NodeId, "quic_ping", start, 100, 100, i => new EchoPacket(false, lossEvery > 0 && i % lossEvery == 0, (long)(owdMs * 1000), (long)(owdMs * 1000))); Assert.True(await agent.SendChunkAsync(udp)); if (tcp) @@ -37,7 +37,7 @@ public sealed class IngestAndQueryTests : IClassFixture long dialerBase = ChunkFactory.DialerUs(0); RawStream info = ChunkFactory.TcpInfo(Enumerable.Range(0, 10).Select(s => (dialerBase + (s * 1_000_000L), (byte)0, 45_000u, (uint)(s * 2), (uint)(s * 400)))); RawChunk tcpChunk = ChunkFactory.Echo( - agent.NodeId, "tcp_echo", start, 200, 50, _ => new EchoPacket(false, false, (long)(owdMs * 1000), (long)(owdMs * 1000)), extraStreams: [info]); + agent.NodeId, "tcp_ping", start, 200, 50, _ => new EchoPacket(false, false, (long)(owdMs * 1000), (long)(owdMs * 1000)), extraStreams: [info]); Assert.True(await agent.SendChunkAsync(tcpChunk)); } } @@ -87,13 +87,13 @@ public sealed class IngestAndQueryTests : IClassFixture // Anchored at now: the last 5 minutes are empty JsonNode atNow = await SeriesAsync(id, "range=5m&step=10s&metrics=loss.total"); - Assert.Null(atNow["summary"]!["samples"]!["udp_echo"]); + Assert.Null(atNow["summary"]!["samples"]!["quic_ping"]); // Anchored at the latest sample: the window ends where the data ends and carries all of it JsonNode anchored = await SeriesAsync(id, "range=5m&step=10s&metrics=loss.total&anchor=latest"); long to = DateTimeOffset.Parse(anchored["to"]!.GetValue(), System.Globalization.CultureInfo.InvariantCulture).ToUnixTimeMilliseconds(); Assert.InRange(to, baseMs + (50 * 1000), baseMs + (80 * 1000)); - Assert.Equal(600, anchored["summary"]!["samples"]!["udp_echo"]!.GetValue()); + Assert.Equal(600, anchored["summary"]!["samples"]!["quic_ping"]!.GetValue()); Assert.False(anchored["live"]!.GetValue()); // 这个窗口是以“最后一个样本”为终点的绝对区间，但线路恢复之后同一个地址的内容就变了：不能被浏览器永久缓存 @@ -102,7 +102,7 @@ public sealed class IngestAndQueryTests : IClassFixture // anchor=now is the default behaviour; anything else is rejected JsonNode explicitNow = await SeriesAsync(id, "range=5m&step=10s&metrics=loss.total&anchor=now"); - Assert.Null(explicitNow["summary"]!["samples"]!["udp_echo"]); + Assert.Null(explicitNow["summary"]!["samples"]!["quic_ping"]); using HttpResponseMessage bad = await _backend.Admin.GetAsync($"/api/public/lines/{_backend.Line(id)}/series?range=5m&anchor=yesterday"); Assert.Equal(System.Net.HttpStatusCode.BadRequest, bad.StatusCode); } @@ -162,8 +162,8 @@ public sealed class IngestAndQueryTests : IClassFixture { minute = await SeriesAsync(agent.NodeId.Value, range + "&step=1m"); // 窗口会随着后续的块到达而重算：等到 18 个块的样本都计入之后再比较 - return minute["source"]!.GetValue() == "1m" && minute["summary"]!["samples"]!["udp_echo"]?.GetValue() == 1800 - && minute["summary"]!["samples"]!["tcp_echo"]?.GetValue() == 3600; + return minute["source"]!.GetValue() == "1m" && minute["summary"]!["samples"]!["quic_ping"]?.GetValue() == 1800 + && minute["summary"]!["samples"]!["tcp_ping"]?.GetValue() == 3600; }); JsonNode hotSummary = hot["summary"]!["metrics"]!; @@ -176,9 +176,9 @@ public sealed class IngestAndQueryTests : IClassFixture $"{id}: hot={hotSummary[id]} persisted={minuteSummary[id]}"); } - Assert.Equal(1800, hot["summary"]!["samples"]!["udp_echo"]!.GetValue()); - Assert.Equal(hot["summary"]!["samples"]!["udp_echo"]!.GetValue(), minute["summary"]!["samples"]!["udp_echo"]!.GetValue()); - Assert.Equal(3600, minute["summary"]!["samples"]!["tcp_echo"]!.GetValue()); + Assert.Equal(1800, hot["summary"]!["samples"]!["quic_ping"]!.GetValue()); + Assert.Equal(hot["summary"]!["samples"]!["quic_ping"]!.GetValue(), minute["summary"]!["samples"]!["quic_ping"]!.GetValue()); + Assert.Equal(3600, minute["summary"]!["samples"]!["tcp_ping"]!.GetValue()); // RTT = 2 × 20 ms，直方图的相对误差 ≤ 4.5% Assert.InRange(minuteSummary["rtt.p95"]!.GetValue(), 40 * 0.955, 40 * 1.045); @@ -234,7 +234,7 @@ public sealed class IngestAndQueryTests : IClassFixture Assert.True(goodScore > 90, $"clean line scored {goodScore}"); Assert.True(badScore < goodScore - 30, $"lossy line scored {badScore}"); Assert.Equal(20.0, lines[badIndex]["summary"]!["metrics"]!["loss.total"]!.GetValue(), 6); - Assert.Equal(1200, lines[goodIndex]["summary"]!["samples"]!["udp_echo"]!.GetValue()); + Assert.Equal(1200, lines[goodIndex]["summary"]!["samples"]!["quic_ping"]!.GetValue()); // 按指标排序：loss.total 升序时好线路在前 JsonNode byLoss = await _backend.GetJsonAsync(_backend.Anonymous, "/api/public/overview?" + range + "&sort=loss.total&order=asc"); @@ -372,12 +372,12 @@ public sealed class IngestAndQueryTests : IClassFixture await using (agent) { string url = $"/api/public/lines/{_backend.Line(agent.NodeId.Value)}/series?from={Iso(baseMs)}&to={Iso(baseMs + Minute)}&step=1m&preset=udp"; - await BackendFixture.EventuallyAsync(async () => (await _backend.GetJsonAsync(_backend.Admin, url))["summary"]!["samples"]!["udp_echo"]?.GetValue() == 600); + await BackendFixture.EventuallyAsync(async () => (await _backend.GetJsonAsync(_backend.Admin, url))["summary"]!["samples"]!["quic_ping"]?.GetValue() == 600); // Agent 重连后补传一个迟到的块（落在同一个分钟窗口里） - RawChunk late = ChunkFactory.Echo(agent.NodeId, "udp_echo", baseMs + 5_000, 100, 10, _ => new EchoPacket(false, false, 20_000, 20_000)); + RawChunk late = ChunkFactory.Echo(agent.NodeId, "quic_ping", baseMs + 5_000, 100, 10, _ => new EchoPacket(false, false, 20_000, 20_000)); Assert.True(await agent.SendChunkAsync(late)); - await BackendFixture.EventuallyAsync(async () => (await _backend.GetJsonAsync(_backend.Admin, url))["summary"]!["samples"]!["udp_echo"]?.GetValue() == 700, because: "the late chunk must re-enter the persistent path"); + await BackendFixture.EventuallyAsync(async () => (await _backend.GetJsonAsync(_backend.Admin, url))["summary"]!["samples"]!["quic_ping"]?.GetValue() == 700, because: "the late chunk must re-enter the persistent path"); } } @@ -394,7 +394,7 @@ public sealed class IngestAndQueryTests : IClassFixture await BackendFixture.EventuallyAsync(async () => { oneMinute = await SeriesAsync(agent.NodeId.Value, range + "&step=1m"); - return oneMinute["summary"]!["samples"]!["udp_echo"]?.GetValue() == 3000; + return oneMinute["summary"]!["samples"]!["quic_ping"]?.GetValue() == 3000; }); // 5 分钟层由降采样把 5 个 1 分钟窗口精确合并而来 @@ -402,7 +402,7 @@ public sealed class IngestAndQueryTests : IClassFixture await BackendFixture.EventuallyAsync(async () => { fiveMinutes = await SeriesAsync(agent.NodeId.Value, range + "&step=5m"); - return fiveMinutes["source"]!.GetValue() == "5m" && fiveMinutes["summary"]!["samples"]!["udp_echo"]?.GetValue() == 3000; + return fiveMinutes["source"]!.GetValue() == "5m" && fiveMinutes["summary"]!["samples"]!["quic_ping"]?.GetValue() == 3000; }); Assert.Equal(oneMinute["summary"]!["metrics"]!["loss.total"]!.GetValue(), fiveMinutes["summary"]!["metrics"]!["loss.total"]!.GetValue(), 9); Assert.Equal(oneMinute["summary"]!["metrics"]!["rtt.p99"]!.GetValue(), fiveMinutes["summary"]!["metrics"]!["rtt.p99"]!.GetValue(), 9); diff --git a/backend/tests/MoShiro.Backend.Tests/Ipv6Tests.cs b/backend/tests/MoShiro.Backend.Tests/Ipv6Tests.cs index 0572b53..ad5e9f9 100644 --- a/backend/tests/MoShiro.Backend.Tests/Ipv6Tests.cs +++ b/backend/tests/MoShiro.Backend.Tests/Ipv6Tests.cs @@ -30,7 +30,7 @@ public sealed class Ipv6Tests : IClassFixture { foreach (string version in new[] { IpVersions.V4, IpVersions.V6 }) { - RawChunk chunk = ChunkFactory.Echo(agent.NodeId, "udp_echo", start + k * 10_000, 100, 100, + RawChunk chunk = ChunkFactory.Echo(agent.NodeId, "quic_ping", start + k * 10_000, 100, 100, i => new EchoPacket(false, version == IpVersions.V6 && i % 2 == 0, 20_000, 20_000)); chunk.Header.DialerId = new NodeId(_backend.DefaultDialerId); chunk.Header.IpVersion = version; @@ -50,7 +50,7 @@ public sealed class Ipv6Tests : IClassFixture await BackendFixture.EventuallyAsync(async () => { series = await _backend.GetJsonAsync(_backend.Admin, $"/api/public/lines/{line}/series?{range}&step={step}&preset=udp"); - return series["summary"]!["samples"]!["udp_echo"]?.GetValue() == 600; + return series["summary"]!["samples"]!["quic_ping"]?.GetValue() == 600; }); Assert.Equal(loss, series["summary"]!["metrics"]!["loss.total"]!.GetValue(), 6); } @@ -99,7 +99,7 @@ public sealed class Ipv6Tests : IClassFixture { JsonNode raw = await _backend.GetJsonAsync(_backend.Admin, "/api/admin/raw/chunks?agent_id=" + agent.Id); return new[] { "line:" + dialer.DialerId, "line:" + dialer.DialerId + ":ipv6" }.All(series => - new[] { "udp_echo", "tcp_echo" }.All(kind => raw["items"]!.AsArray().Any(c => c!["series"]!.GetValue() == series && c["kind"]!.GetValue() == kind))); + new[] { "quic_ping", "tcp_ping" }.All(kind => raw["items"]!.AsArray().Any(c => c!["series"]!.GetValue() == series && c["kind"]!.GetValue() == kind))); }, TimeSpan.FromSeconds(45)); string v4 = agent.Id + "~" + dialer.DialerId; @@ -110,8 +110,8 @@ public sealed class Ipv6Tests : IClassFixture { JsonNode series = await _backend.GetJsonAsync(_backend.Admin, $"/api/public/lines/{line}/series?range=1m&step={step}&preset=udp,tcp"); Assert.Equal(family, series["ip_version"]!.GetValue()); - Assert.True(series["summary"]!["samples"]!["udp_echo"]!.GetValue() > 0, series.ToJsonString()); - Assert.True(series["summary"]!["samples"]!["tcp_echo"]!.GetValue() > 0, series.ToJsonString()); + Assert.True(series["summary"]!["samples"]!["quic_ping"]!.GetValue() > 0, series.ToJsonString()); + Assert.True(series["summary"]!["samples"]!["tcp_ping"]!.GetValue() > 0, series.ToJsonString()); } } diff --git a/backend/tests/MoShiro.Backend.Tests/LiveTests.cs b/backend/tests/MoShiro.Backend.Tests/LiveTests.cs index d22581b..cf9b596 100644 --- a/backend/tests/MoShiro.Backend.Tests/LiveTests.cs +++ b/backend/tests/MoShiro.Backend.Tests/LiveTests.cs @@ -34,9 +34,9 @@ public sealed class LiveTests : IClassFixture await sse.WaitForAsync(e => e.Name == "series" && e.Data!["agent_id"]!.GetValue() == id); // 新块到达之后：推送新的 1s 桶（列式）和最新的汇总 - RawChunk chunk = ChunkFactory.Echo(agent.NodeId, "udp_echo", Now() - 10_000, 100, 100, i => new EchoPacket(false, i % 10 == 0, 20_000, 20_000)); + RawChunk chunk = ChunkFactory.Echo(agent.NodeId, "quic_ping", Now() - 10_000, 100, 100, i => new EchoPacket(false, i % 10 == 0, 20_000, 20_000)); Assert.True(await agent.SendChunkAsync(chunk)); - SseEvent series = await sse.WaitForAsync(e => e.Name == "series" && e.Data!["summary"]!["samples"]!["udp_echo"]?.GetValue() >= 100); + SseEvent series = await sse.WaitForAsync(e => e.Name == "series" && e.Data!["summary"]!["samples"]!["quic_ping"]?.GetValue() >= 100); Assert.True(series.Data!["live"]!.GetValue()); Assert.Equal("1s", series.Data["step"]!.GetValue()); Assert.Equal(series.Data["t"]!.AsArray().Count, series.Data["values"]!["loss.total"]!.AsArray().Count); diff --git a/backend/tests/MoShiro.Backend.Tests/OnDemandTestsTests.cs b/backend/tests/MoShiro.Backend.Tests/OnDemandTestsTests.cs index 5c50c98..889d7b2 100644 --- a/backend/tests/MoShiro.Backend.Tests/OnDemandTestsTests.cs +++ b/backend/tests/MoShiro.Backend.Tests/OnDemandTestsTests.cs @@ -82,7 +82,7 @@ public sealed class OnDemandTestsTests : IClassFixture TestRunParams run = agent.TestRuns.Single(); Assert.Equal(testId, run.TestId.Value); Assert.Equal([10d, 25d, 50d, 100d], run.Phases.Select(p => p.Params["mbps"].GetDouble()).ToArray()); - Assert.All(run.Phases, p => Assert.Equal("udp_train", p.Kind)); + Assert.All(run.Phases, p => Assert.Equal("quic_stream", p.Kind)); Assert.All(run.Phases, p => Assert.Equal(1250, p.Params["size"].GetInt64())); // 同一条线路同时只能跑一个测试 @@ -227,7 +227,7 @@ public sealed class OnDemandTestsTests : IClassFixture RawChunk chunk = ChunkFactory.UdpEchoClean(agent.NodeId, now - 120_000); Assert.True(await agent.SendChunkAsync(chunk)); - JsonNode list = await _backend.GetJsonAsync(_backend.Admin, $"/api/admin/raw/chunks?agent_id={agent.NodeId.Value}&kind=udp_echo"); + JsonNode list = await _backend.GetJsonAsync(_backend.Admin, $"/api/admin/raw/chunks?agent_id={agent.NodeId.Value}&kind=quic_ping"); JsonNode item = list["items"]!.AsArray().Single()!; Assert.Equal(chunk.Header.ChunkId.ToString(), item["chunk_id"]!.GetValue()); Assert.Equal("line:" + _backend.DefaultDialerId, item["series"]!.GetValue()); diff --git a/backend/tests/MoShiro.Backend.Tests/OperationsAndSecurityTests.cs b/backend/tests/MoShiro.Backend.Tests/OperationsAndSecurityTests.cs index c49e4b6..fab7802 100644 --- a/backend/tests/MoShiro.Backend.Tests/OperationsAndSecurityTests.cs +++ b/backend/tests/MoShiro.Backend.Tests/OperationsAndSecurityTests.cs @@ -42,7 +42,7 @@ public sealed class OperationsAndSecurityTests : IClassFixture await using AgentSimulator agent = await _backend.ConnectAgentAsync(); for (int k = 0; k < 12; k++) { - Assert.True(await agent.SendChunkAsync(ChunkFactory.Echo(agent.NodeId, "udp_echo", baseMs + (k * 10_000L), 100, 100, _ => new EchoPacket(false, false, 30_000, 30_000)))); + Assert.True(await agent.SendChunkAsync(ChunkFactory.Echo(agent.NodeId, "quic_ping", baseMs + (k * 10_000L), 100, 100, _ => new EchoPacket(false, false, 30_000, 30_000)))); } string url = $"/api/public/lines/{_backend.Line(agent.NodeId.Value)}/series?from={Iso(baseMs)}&to={Iso(baseMs + (2 * Minute))}&step=1m&metrics=rtt.p95,loss.total"; @@ -50,7 +50,7 @@ public sealed class OperationsAndSecurityTests : IClassFixture await BackendFixture.EventuallyAsync(async () => { before = await _backend.GetJsonAsync(_backend.Admin, url); - return before["summary"]!["samples"]!["udp_echo"]?.GetValue() == 1200 && before["source"]!.GetValue() == "1m"; + return before["summary"]!["samples"]!["quic_ping"]?.GetValue() == 1200 && before["source"]!.GetValue() == "1m"; }); double rtt = before["summary"]!["metrics"]!["rtt.p95"]!.GetValue(); Assert.InRange(rtt, 60 * 0.955, 60 * 1.045); diff --git a/backend/tests/MoShiro.Backend.Tests/PersistenceTests.cs b/backend/tests/MoShiro.Backend.Tests/PersistenceTests.cs index 87be2b7..e986bb4 100644 --- a/backend/tests/MoShiro.Backend.Tests/PersistenceTests.cs +++ b/backend/tests/MoShiro.Backend.Tests/PersistenceTests.cs @@ -21,14 +21,14 @@ public sealed class PersistenceTests : IAsyncLifetime string id = agent.NodeId.Value; for (int k = 0; k < 12; k++) { - Assert.True(await agent.SendChunkAsync(ChunkFactory.Echo(agent.NodeId, "udp_echo", baseMs + (k * 10_000L), 100, 100, i => new EchoPacket(false, i % 4 == 0, 20_000, 20_000)))); + Assert.True(await agent.SendChunkAsync(ChunkFactory.Echo(agent.NodeId, "quic_ping", baseMs + (k * 10_000L), 100, 100, i => new EchoPacket(false, i % 4 == 0, 20_000, 20_000)))); } string from = DateTimeOffset.FromUnixTimeMilliseconds(baseMs).UtcDateTime.ToString("yyyy-MM-dd'T'HH:mm:ss.fff'Z'"); string to = DateTimeOffset.FromUnixTimeMilliseconds(baseMs + (2 * minute)).UtcDateTime.ToString("yyyy-MM-dd'T'HH:mm:ss.fff'Z'"); string url = $"/api/public/lines/{_backend.Line(id)}/series?from={from}&to={to}&metrics=loss.total,rtt.p95"; JsonNode before = await _backend.GetJsonAsync(_backend.Admin, url + "&step=1m"); - await BackendFixture.EventuallyAsync(async () => (await _backend.GetJsonAsync(_backend.Admin, url + "&step=1m"))["summary"]!["samples"]!["udp_echo"]?.GetValue() == 1200); + await BackendFixture.EventuallyAsync(async () => (await _backend.GetJsonAsync(_backend.Admin, url + "&step=1m"))["summary"]!["samples"]!["quic_ping"]?.GetValue() == 1200); before = await _backend.GetJsonAsync(_backend.Admin, url + "&step=1m"); await agent.DisposeAsync(); @@ -51,7 +51,7 @@ public sealed class PersistenceTests : IAsyncLifetime await BackendFixture.EventuallyAsync(async () => { hot = await _backend.GetJsonAsync(admin, url + "&step=1s"); - return hot["source"]!.GetValue() == "hot" && hot["summary"]!["samples"]!["udp_echo"]?.GetValue() == 1200; + return hot["source"]!.GetValue() == "hot" && hot["summary"]!["samples"]!["quic_ping"]?.GetValue() == 1200; }); Assert.Equal(25.0, hot["summary"]!["metrics"]!["loss.total"]!.GetValue(), 6); diff --git a/backend/tests/MoShiro.Backend.Tests/ScheduleTests.cs b/backend/tests/MoShiro.Backend.Tests/ScheduleTests.cs index c64713c..68bed87 100644 --- a/backend/tests/MoShiro.Backend.Tests/ScheduleTests.cs +++ b/backend/tests/MoShiro.Backend.Tests/ScheduleTests.cs @@ -72,8 +72,8 @@ public sealed class ScheduleTests : IClassFixture Assert.Equal(HttpStatusCode.OK, response.StatusCode); JsonArray items = JsonNode.Parse(await response.Content.ReadAsStringAsync())!["items"]!.AsArray(); string[] kinds = [.. items.Select(i => i!["kind"]!.GetValue())]; - Assert.Contains("connect", kinds); - Assert.Contains("udp_echo", kinds); - Assert.Contains("tcp_echo", kinds); + Assert.Contains("tcp_connect", kinds); + Assert.Contains("quic_ping", kinds); + Assert.Contains("tcp_ping", kinds); } } diff --git a/backend/tests/MoShiro.Backend.Tests/StorageTests.cs b/backend/tests/MoShiro.Backend.Tests/StorageTests.cs index f30a9f0..fcdf763 100644 --- a/backend/tests/MoShiro.Backend.Tests/StorageTests.cs +++ b/backend/tests/MoShiro.Backend.Tests/StorageTests.cs @@ -48,8 +48,8 @@ public sealed class StorageTests : IDisposable public async Task Raw_AppendReadAndDedupe() { var raw = new SqliteRawStore(_db); - RawChunkAppend a = Chunk(s_agent, "udp_echo", 1_000, "2026-01-01"); - RawChunkAppend b = Chunk(s_agent, "udp_echo", 11_000, "2026-01-01"); + RawChunkAppend a = Chunk(s_agent, "quic_ping", 1_000, "2026-01-01"); + RawChunkAppend b = Chunk(s_agent, "quic_ping", 11_000, "2026-01-01"); Assert.True(await raw.AppendAsync(a, default)); Assert.True(await raw.AppendAsync(b, default)); @@ -57,7 +57,7 @@ public sealed class StorageTests : IDisposable IReadOnlyList metas = await raw.FindAsync(new RawQuery(s_agent), default); Assert.Equal(2, metas.Count); - Assert.Equal("2026-01-01/agt_storage/udp_echo.seg", metas[0].SegmentPath); + Assert.Equal("2026-01-01/agt_storage/quic_ping.seg", metas[0].SegmentPath); Assert.Equal(0, metas[0].Offset); Assert.Equal(metas[0].Length, metas[1].Offset); // 同一个段文件里依次追加 Assert.Equal(a.Bytes.ToArray(), await raw.ReadAsync(metas[0], default)); @@ -72,7 +72,7 @@ public sealed class StorageTests : IDisposable var raw = new SqliteRawStore(_db); var evil = new NodeId("../../../outside"); Assert.True(await raw.AppendAsync(Chunk(evil, "../etc", 1, "2026-01-01"), default)); - Assert.True(await raw.AppendAsync(Chunk(s_agent, "udp_echo", 2, "2026-01-01", test: new TestId("..\\..\\x"), phase: 0), default)); + Assert.True(await raw.AppendAsync(Chunk(s_agent, "quic_ping", 2, "2026-01-01", test: new TestId("..\\..\\x"), phase: 0), default)); foreach (string file in Directory.EnumerateFiles(_dir, "*.seg", SearchOption.AllDirectories)) { @@ -80,7 +80,7 @@ public sealed class StorageTests : IDisposable } Assert.False(File.Exists(Path.Combine(_dir, "outside"))); - await Assert.ThrowsAsync(async () => await raw.AppendAsync(Chunk(s_agent, "udp_echo", 3, "../2026"), default)); + await Assert.ThrowsAsync(async () => await raw.AppendAsync(Chunk(s_agent, "quic_ping", 3, "../2026"), default)); } [Fact] @@ -90,19 +90,19 @@ public sealed class StorageTests : IDisposable var testId = new TestId("tst_one"); for (int i = 0; i < 6; i++) { - await raw.AppendAsync(Chunk(s_agent, "udp_echo", i * 10_000L, "2026-01-01"), default); + await raw.AppendAsync(Chunk(s_agent, "quic_ping", i * 10_000L, "2026-01-01"), default); } - await raw.AppendAsync(Chunk(s_agent, "tcp_echo", 5_000, "2026-01-01"), default); - await raw.AppendAsync(Chunk(new NodeId("agt_other"), "udp_echo", 5_000, "2026-01-01"), default); - await raw.AppendAsync(Chunk(s_agent, "udp_train", 0, "2026-01-01", test: testId, phase: 1), default); + await raw.AppendAsync(Chunk(s_agent, "tcp_ping", 5_000, "2026-01-01"), default); + await raw.AppendAsync(Chunk(new NodeId("agt_other"), "quic_ping", 5_000, "2026-01-01"), default); + await raw.AppendAsync(Chunk(s_agent, "quic_stream", 0, "2026-01-01", test: testId, phase: 1), default); Assert.Equal(8, (await raw.FindAsync(new RawQuery(s_agent), default)).Count); - Assert.Equal(6, (await raw.FindAsync(new RawQuery(s_agent, Kind: "udp_echo"), default)).Count); + Assert.Equal(6, (await raw.FindAsync(new RawQuery(s_agent, Kind: "quic_ping"), default)).Count); Assert.Single(await raw.FindAsync(new RawQuery(TestId: testId), default)); - Assert.Equal(3, (await raw.FindAsync(new RawQuery(s_agent, Kind: "udp_echo", FromMs: 10_000, ToMs: 40_000), default)).Count); - Assert.Equal(2, (await raw.FindAsync(new RawQuery(s_agent, Kind: "udp_echo", Limit: 2), default)).Count); - Assert.Equal(4, (await raw.FindAsync(new RawQuery(s_agent, Kind: "udp_echo", AfterStartTs: 20_000), default)).Count); + Assert.Equal(3, (await raw.FindAsync(new RawQuery(s_agent, Kind: "quic_ping", FromMs: 10_000, ToMs: 40_000), default)).Count); + Assert.Equal(2, (await raw.FindAsync(new RawQuery(s_agent, Kind: "quic_ping", Limit: 2), default)).Count); + Assert.Equal(4, (await raw.FindAsync(new RawQuery(s_agent, Kind: "quic_ping", AfterStartTs: 20_000), default)).Count); IReadOnlyList<(NodeId AgentId, string Series)> series = await raw.ListSeriesAsync(0, 100_000, default); Assert.Equal(2, series.Count); // 测试的块不算持续序列 @@ -112,9 +112,9 @@ public sealed class StorageTests : IDisposable public async Task Raw_RecoverTruncatesAnUnindexedTail_AndLaterAppendsStayConsistent() { var raw = new SqliteRawStore(_db); - RawChunkAppend first = Chunk(s_agent, "udp_echo", 1, "2026-01-01"); + RawChunkAppend first = Chunk(s_agent, "quic_ping", 1, "2026-01-01"); await raw.AppendAsync(first, default); - string segment = Directory.EnumerateFiles(_dir, "udp_echo.seg", SearchOption.AllDirectories).Single(); + string segment = Directory.EnumerateFiles(_dir, "quic_ping.seg", SearchOption.AllDirectories).Single(); long indexedEnd = new FileInfo(segment).Length; // 追加之后、写索引之前崩溃：段文件尾部多了没有索引记录的字节 @@ -124,7 +124,7 @@ public sealed class StorageTests : IDisposable await raw.RecoverAsync(default); Assert.Equal(indexedEnd, new FileInfo(segment).Length); - RawChunkAppend second = Chunk(s_agent, "udp_echo", 2, "2026-01-01"); + RawChunkAppend second = Chunk(s_agent, "quic_ping", 2, "2026-01-01"); await raw.AppendAsync(second, default); IReadOnlyList metas = await raw.FindAsync(new RawQuery(s_agent), default); Assert.Equal(first.Bytes.ToArray(), await raw.ReadAsync(metas[0], default)); @@ -137,10 +137,10 @@ public sealed class StorageTests : IDisposable { var raw = new SqliteRawStore(_db); var testId = new TestId("tst_old"); - await raw.AppendAsync(Chunk(s_agent, "udp_echo", 1, "2026-01-01"), default); - await raw.AppendAsync(Chunk(s_agent, "udp_echo", 2, "2026-03-01"), default); - await raw.AppendAsync(Chunk(s_agent, "udp_train", 3, "2026-01-01", test: testId, phase: 0), default); - await raw.AppendAsync(Chunk(s_agent, "udp_train", 4, "2026-02-01", test: new TestId("tst_mid"), phase: 0), default); + await raw.AppendAsync(Chunk(s_agent, "quic_ping", 1, "2026-01-01"), default); + await raw.AppendAsync(Chunk(s_agent, "quic_ping", 2, "2026-03-01"), default); + await raw.AppendAsync(Chunk(s_agent, "quic_stream", 3, "2026-01-01", test: testId, phase: 0), default); + await raw.AppendAsync(Chunk(s_agent, "quic_stream", 4, "2026-02-01", test: new TestId("tst_mid"), phase: 0), default); // 持续任务保留到 2026-02-01 之后，测试保留到 2026-01-15 之后 int removed = await raw.DeleteExpiredAsync(new DateOnly(2026, 2, 1), new DateOnly(2026, 1, 15), default); @@ -365,6 +365,44 @@ public sealed class StorageTests : IDisposable Assert.Equal(200, (await settings.GetAllAsync(default)).Count); } + [Fact] + public void Schema_UpgradesVersion1_ByRenamingTheKinds() + { + string dir = Path.Combine(Path.GetTempPath(), "moshiro-schema-" + Guid.NewGuid().ToString("N")); + Directory.CreateDirectory(dir); + string path = Path.Combine(dir, "moshiro.db"); + try + { + using (var db = new SqliteDatabase(dir)) + { + } + + SqliteConnection.ClearAllPools(); + using (var connection = new SqliteConnection($"Data Source={path}")) + { + connection.Open(); + connection.Exec("INSERT INTO tasks (id, scope, kind, schedule, updated_at) VALUES ('tsk_a', 'line', 'udp_echo', 'continuous', 1), ('tsk_b', 'line', 'connect', 'continuous', 1)"); + connection.Exec("PRAGMA user_version = 1"); + } + + SqliteConnection.ClearAllPools(); + using (var db = new SqliteDatabase(dir)) + { + } + + SqliteConnection.ClearAllPools(); + using var check = new SqliteConnection($"Data Source={path}"); + check.Open(); + Assert.Equal(2, check.Scalar("PRAGMA user_version")); + Assert.Equal("quic_ping,tcp_connect", string.Join(',', new[] { "tsk_a", "tsk_b" }.Select(id => check.Scalar($"SELECT kind FROM tasks WHERE id = '{id}'")))); + } + finally + { + SqliteConnection.ClearAllPools(); + Directory.Delete(dir, recursive: true); + } + } + [Fact] public void Schema_RefusesADatabaseFromANewerBuild() { diff --git a/backend/tests/MoShiro.TestSupport/AgentSimulator.cs b/backend/tests/MoShiro.TestSupport/AgentSimulator.cs index 3fbc8c8..5d4d831 100644 --- a/backend/tests/MoShiro.TestSupport/AgentSimulator.cs +++ b/backend/tests/MoShiro.TestSupport/AgentSimulator.cs @@ -109,7 +109,7 @@ public sealed class AgentSimulator : IAsyncDisposable ["tcp_port"] = listens ? 7443 : 0, ["quic_ports"] = listens ? new JsonArray(7443) : new JsonArray(), ["public_host"] = listens ? publicHost : "", - ["kinds"] = new JsonArray((kinds ?? ["udp_echo", "tcp_echo", "udp_train", "connect", "hold", "tcp_bulk", "icmp_echo"]).Select(k => (JsonNode?)JsonValue.Create(k)).ToArray()), + ["kinds"] = new JsonArray((kinds ?? ["quic_ping", "tcp_ping", "quic_stream", "tcp_connect", "tls_hold", "tcp_transfer", "icmp_ping"]).Select(k => (JsonNode?)JsonValue.Create(k)).ToArray()), }, }; @@ -169,7 +169,7 @@ public sealed class AgentSimulator : IAsyncDisposable var hello = new HelloParams { NodeVersion = "0.0.1-sim", - Capabilities = [MoShiroCapabilities.Base, MoShiroCapabilities.KindsPrefix + string.Join(',', kinds ?? ["udp_echo", "tcp_echo", "udp_train", "connect", "hold", "tcp_bulk", "icmp_echo"])], + Capabilities = [MoShiroCapabilities.Base, MoShiroCapabilities.KindsPrefix + string.Join(',', kinds ?? ["quic_ping", "tcp_ping", "quic_stream", "tcp_connect", "tls_hold", "tcp_transfer", "icmp_ping"])], NowUtcMs = TimeProvider.System.GetUtcNow().ToUnixTimeMilliseconds(), Meta = meta, }; diff --git a/backend/tests/MoShiro.TestSupport/ChunkFactory.cs b/backend/tests/MoShiro.TestSupport/ChunkFactory.cs index 9011337..aa0bb2c 100644 --- a/backend/tests/MoShiro.TestSupport/ChunkFactory.cs +++ b/backend/tests/MoShiro.TestSupport/ChunkFactory.cs @@ -71,13 +71,13 @@ public static class ChunkFactory return new RawChunk(Header(agent, kind, startUtcMs, durationMs, parameters), streams); } - /// 10 Hz 的 udp_echo，没有丢包，RTT = 2 × owdMs。 + /// 10 Hz 的 quic_ping，没有丢包，RTT = 2 × owdMs。 public static RawChunk UdpEchoClean(NodeId agent, long startUtcMs, int count = 100, double owdMs = 20) => - Echo(agent, "udp_echo", startUtcMs, count, 100, _ => new EchoPacket(false, false, (long)(owdMs * 1000), (long)(owdMs * 1000))); + Echo(agent, "quic_ping", startUtcMs, count, 100, _ => new EchoPacket(false, false, (long)(owdMs * 1000), (long)(owdMs * 1000))); public static RawStream TcpInfo(IEnumerable<(long TsUs, byte Side, uint SrttUs, uint RetransTotal, uint SegsOut)> samples) { - var builder = new StreamBuilder(Kinds.TcpEcho.FindStream("tcp_info")!); + var builder = new StreamBuilder(Kinds.TcpPing.FindStream("tcp_info")!); foreach ((long ts, byte side, uint srtt, uint retrans, uint segs) in samples) { builder.Add(ts, side, srtt, srtt / 4, 10, retrans, segs, segs * 100L, null); @@ -88,7 +88,7 @@ public static class ChunkFactory public static RawChunk Connect(NodeId agent, long startUtcMs, IEnumerable<(long TcpUs, long? TlsUs, ConnectOutcomeCode Outcome)> attempts) { - var builder = new StreamBuilder(Kinds.Connect.FindStream("conn")!); + var builder = new StreamBuilder(Kinds.TcpConnect.FindStream("conn")!); int i = 0; foreach ((long tcp, long? tls, ConnectOutcomeCode outcome) in attempts) { @@ -96,16 +96,16 @@ public static class ChunkFactory i++; } - return new RawChunk(Header(agent, "connect", startUtcMs), [builder.Build()]); + return new RawChunk(Header(agent, "tcp_connect", startUtcMs), [builder.Build()]); } - /// udp_train：Agent 以固定间隔发包，Dialer 记录收到的时刻； 为 N 时每 N 个包丢一个。 + /// quic_stream：Agent 以固定间隔发包，Dialer 记录收到的时刻； 为 N 时每 N 个包丢一个。 public static RawChunk Train(NodeId agent, long startUtcMs, double mbps, int phase, int seconds = 10, int lossEvery = 0, long owdUs = 20_000) { const int size = 1250; int packets = (int)(mbps * 1_000_000 / 8 / size * seconds); double intervalUs = seconds * 1_000_000.0 / packets; - var pkt = new StreamBuilder(Kinds.UdpTrain.FindStream("pkt")!); + var pkt = new StreamBuilder(Kinds.QuicStream.FindStream("pkt")!); for (int i = 0; i < packets; i++) { long sendAgent = AgentMonoAnchorUs + (long)(i * intervalUs); @@ -119,10 +119,10 @@ public static class ChunkFactory ["mbps"] = JsonSerializer.SerializeToElement(mbps, FactoryJson.Default.Double), ["duration_s"] = JsonSerializer.SerializeToElement((double)seconds, FactoryJson.Default.Double), }; - RawChunkHeader header = Header(agent, "udp_train", startUtcMs, seconds * 1000, parameters); + RawChunkHeader header = Header(agent, "quic_stream", startUtcMs, seconds * 1000, parameters); header.Phase = phase; header.TestId = new TestId("tst_test"); - return new RawChunk(header, [pkt.Build(), new StreamBuilder(Kinds.UdpTrain.FindStream("step")!).Build()]); + return new RawChunk(header, [pkt.Build(), new StreamBuilder(Kinds.QuicStream.FindStream("step")!).Build()]); } } diff --git a/backend/tests/MoShiro.TestSupport/DialerSimulator.cs b/backend/tests/MoShiro.TestSupport/DialerSimulator.cs index 64ae5f0..84404b7 100644 --- a/backend/tests/MoShiro.TestSupport/DialerSimulator.cs +++ b/backend/tests/MoShiro.TestSupport/DialerSimulator.cs @@ -138,7 +138,7 @@ public sealed class DialerSimulator : IAsyncDisposable } } - /// 对一个 udp_echo 线路任务发 个回显包，等回复，然后交出一个列块。返回交给 Agent 的列块的序号范围。 + /// 对一个 quic_ping 线路任务发 个回显包，等回复，然后交出一个列块。返回交给 Agent 的列块的序号范围。 public async Task<(int Sent, int Received)> RunUdpEchoAsync(LineTask task, int count, TimeSpan interval, CancellationToken ct = default) { int size = (int)task.Params["size"].GetInt64(); @@ -166,12 +166,12 @@ public sealed class DialerSimulator : IAsyncDisposable msg.Add(seq, size, 1, t1, collector.Received.TryGetValue(seq, out long t4) ? t4 : null); } - RawChunkHeader header = Header("udp_echo", task.Id, null, null, startUtc, (int)Math.Max(1, count * interval.TotalMilliseconds), task.Params); + RawChunkHeader header = Header("quic_ping", task.Id, null, null, startUtc, (int)Math.Max(1, count * interval.TotalMilliseconds), task.Params); await ShipAsync(new RawChunk(header, [msg.Build()]), ct: ct); return (count, collector.Received.Count); } - /// 对一个 tcp_echo 线路任务建立一条 TCP（可选 TLS）连接，发 条回显消息，然后交出一个列块。 + /// 对一个 tcp_ping 线路任务建立一条 TCP（可选 TLS）连接，发 条回显消息，然后交出一个列块。 public async Task<(int Sent, int Received)> RunTcpEchoAsync(LineTask task, int tcpPort, int count, TimeSpan interval, bool tls, CancellationToken ct = default) { using var timeout = CancellationTokenSource.CreateLinkedTokenSource(ct); @@ -211,7 +211,7 @@ public sealed class DialerSimulator : IAsyncDisposable await Task.Delay(interval, ct); } - RawChunkHeader header = Header("tcp_echo", task.Id, null, null, startUtc, (int)Math.Max(1, count * interval.TotalMilliseconds), task.Params); + RawChunkHeader header = Header("tcp_ping", task.Id, null, null, startUtc, (int)Math.Max(1, count * interval.TotalMilliseconds), task.Params); await ShipAsync(new RawChunk(header, [msg.Build()]), ct: ct); return (count, received); } @@ -394,7 +394,7 @@ public sealed class DialerSimulator : IAsyncDisposable } _activePhases[start.StreamId] = start; - if (start.Kind == "udp_train") + if (start.Kind == "quic_stream") { _trains[start.StreamId] = new TrainCollector { StartUtcMs = TimeProvider.System.GetUtcNow().ToUnixTimeMilliseconds() }; _quic!.Register(start.StreamId, OnDatagram); @@ -424,7 +424,7 @@ public sealed class DialerSimulator : IAsyncDisposable return 0; } - if (start.Kind == "udp_train" && _trains.TryRemove(end.StreamId, out TrainCollector? train)) + if (start.Kind == "quic_stream" && _trains.TryRemove(end.StreamId, out TrainCollector? train)) { await Task.Delay(200, ct); // 最后一批包还在路上 // 单块的记录数有上限（ProtocolLimits.MaxRecordsPerChunk）：按序号范围切成多个列块，Agent 一侧按同样的范围取行 @@ -443,12 +443,12 @@ public sealed class DialerSimulator : IAsyncDisposable var streams = new List { pkt.Build() }; if (fragments == 0) { - var step = new StreamBuilder(Kinds.UdpTrain.FindStream("step")!); + var step = new StreamBuilder(Kinds.QuicStream.FindStream("step")!); step.Add(0, (long)start.Params["mbps"].GetDouble(), NowUs() - (start.DurationSeconds * 1_000_000L), NowUs()); streams.Add(step.Build()); } - RawChunkHeader header = Header("udp_train", null, start.TestId, start.Phase, train.StartUtcMs, start.DurationSeconds * 1000, start.Params); + RawChunkHeader header = Header("quic_stream", null, start.TestId, start.Phase, train.StartUtcMs, start.DurationSeconds * 1000, start.Params); header.Params[FragmentKeys.SeqFrom] = JsonSerializer.SerializeToElement(from); header.Params[FragmentKeys.SeqTo] = JsonSerializer.SerializeToElement(to); // 不能在这里等确认：确认是 Agent 的通知，由同一个入站队列顺序处理，而当前正处在 phase_end 的处理函数里