diff --git a/backend/agent/Directory.Build.props b/backend/agent/Directory.Build.props
index 17e23af..8ddfaef 100644
--- a/backend/agent/Directory.Build.props
+++ b/backend/agent/Directory.Build.props
@@ -8,7 +8,7 @@
false
false
false
- false
+ true
false
true
false
diff --git a/backend/agent/src/MoShiro.Agent/AgentApp.cs b/backend/agent/src/MoShiro.Agent/AgentApp.cs
index fa8fc38..b80ffdf 100644
--- a/backend/agent/src/MoShiro.Agent/AgentApp.cs
+++ b/backend/agent/src/MoShiro.Agent/AgentApp.cs
@@ -74,10 +74,16 @@ public sealed class AgentApp : IDisposable, IAsyncDisposable
_ipv6Supported = _listening ? bind.AddressFamily == System.Net.Sockets.AddressFamily.InterNetworkV6 : null;
_admission = new AgentAdmission(_streams);
- _datagrams = new QuicDatagramTransport(_sessions, _admission, _clock, _counters.Data, log, bind, config.Listen.QuicPorts, identity.Certificate);
+ QuicSettings quic = config.Quic.ToSettings();
+ if (quic.Validate() is { } invalid)
+ {
+ throw new InvalidOperationException("agent.json quic: " + invalid);
+ }
+
+ _datagrams = new QuicDatagramTransport(_sessions, _admission, _clock, _counters.Data, log, bind, config.Listen.QuicPorts, identity.Certificate, quic);
// 预握手敲门：控制端口和数据端口共用一个关卡（限速和去重一起算）；名册就是期望状态里的 peers
_trust = TrustAnchor.Load(config.Fleet.IdentityDir, log);
- _dialer = new DialerHost(new DialEnv(identity.PeerIdentity, identity.NodeId.Value, config.Limits, new DialDirectory(new ConfigStore(dataDirectory), dataDirectory, log, _trust), _clock, time, log), time, log);
+ _dialer = new DialerHost(new DialEnv(identity.PeerIdentity, identity.NodeId.Value, config.Limits, quic, new DialDirectory(new ConfigStore(dataDirectory), dataDirectory, log, _trust), _clock, time, log), time, log);
_knock = new KnockGate(identity.PeerIdentity, RosterKeys, time);
_tcp = new TcpAcceptor(bind, config.Listen.TcpPorts, identity.Certificate, _sessions, _counters.Data, _knock);
diff --git a/backend/agent/src/MoShiro.Agent/AgentConfig.cs b/backend/agent/src/MoShiro.Agent/AgentConfig.cs
index 16d4447..6754f55 100644
--- a/backend/agent/src/MoShiro.Agent/AgentConfig.cs
+++ b/backend/agent/src/MoShiro.Agent/AgentConfig.cs
@@ -15,6 +15,9 @@ public sealed class AgentConfig
[JsonPropertyName("listen")]
public ListenSection Listen { get; set; } = new();
+ [JsonPropertyName("quic")]
+ public QuicSection Quic { get; set; } = new();
+
[JsonPropertyName("limits")]
public LimitsSection Limits { get; set; } = new();
@@ -66,6 +69,27 @@ public sealed class ListenSection
public List TcpPorts { get; set; } = [7444];
}
+/// 数据通道 QUIC 连接的传输参数；不写的项用 MsQuic 默认值。监听和拨出的连接都用这里的值。
+public sealed class QuicSection
+{
+ [JsonPropertyName("min_mtu")]
+ public int? MinMtu { get; set; }
+
+ [JsonPropertyName("max_mtu")]
+ public int? MaxMtu { get; set; }
+
+ [JsonPropertyName("pacing")]
+ public bool? Pacing { get; set; }
+
+ [JsonPropertyName("idle_timeout_ms")]
+ public int? IdleTimeoutMs { get; set; }
+
+ [JsonPropertyName("keep_alive_ms")]
+ public int? KeepAliveMs { get; set; }
+
+ public QuicSettings ToSettings() => new() { MinMtu = MinMtu, MaxMtu = MaxMtu, Pacing = Pacing, IdleTimeoutMs = IdleTimeoutMs, KeepAliveMs = KeepAliveMs };
+}
+
public sealed class LimitsSection
{
[JsonPropertyName("max_total_mbps")]
diff --git a/backend/agent/src/MoShiro.Agent/DataPlane/Registries.cs b/backend/agent/src/MoShiro.Agent/DataPlane/Registries.cs
index cabd503..5c061ab 100644
--- a/backend/agent/src/MoShiro.Agent/DataPlane/Registries.cs
+++ b/backend/agent/src/MoShiro.Agent/DataPlane/Registries.cs
@@ -217,8 +217,8 @@ public sealed class StreamInfo
KindSpec spec = Kinds.Find(kind)!;
DeclaredPps = kind switch
{
- "udp_echo" or "tcp_echo" => Kinds.PeakRateHz(kind, parameters),
- "udp_train" => parameters.GetNumber("mbps") * 1_000_000 / 8 / Math.Max(1, parameters.GetInt("size")),
+ "quic_ping" or "tcp_ping" => Kinds.PeakRateHz(kind, parameters),
+ "quic_stream" => parameters.GetNumber("mbps") * 1_000_000 / 8 / Math.Max(1, parameters.GetInt("size")),
_ => 20,
};
Port = spec.Params.Any(p => p.Name == "port") ? (int)parameters.GetInt("port") : 0;
diff --git a/backend/agent/src/MoShiro.Agent/Dial/DialConnection.cs b/backend/agent/src/MoShiro.Agent/Dial/DialConnection.cs
index 8161c2c..5d6dd6a 100644
--- a/backend/agent/src/MoShiro.Agent/Dial/DialConnection.cs
+++ b/backend/agent/src/MoShiro.Agent/Dial/DialConnection.cs
@@ -11,7 +11,7 @@ using MoShiro.Protocol;
namespace MoShiro.Agent;
/// 所有连接共用的东西。
-public sealed record DialEnv(NodeIdentity Identity, string NodeId, LimitsSection Limits, DialDirectory Directory, MonoClock Clock, TimeProvider Time, INodeLog Log);
+public sealed record DialEnv(NodeIdentity Identity, string NodeId, LimitsSection Limits, QuicSettings Quic, DialDirectory Directory, MonoClock Clock, TimeProvider Time, INodeLog Log);
///
/// 与一个 Agent 的点对点连接，分两层：
@@ -170,7 +170,8 @@ public sealed class DialConnection : IDisposable
new DataPlaneCounters(),
_attemptLog,
new IPEndPoint(remote.Address, quicPort),
- agentKey);
+ agentKey,
+ _env.Quic);
Volatile.Write(ref _activeDatagrams, datagrams);
using CancellationTokenSource cts = CancellationTokenSource.CreateLinkedTokenSource(ct);
diff --git a/backend/agent/src/MoShiro.Agent/Dial/DialSession.cs b/backend/agent/src/MoShiro.Agent/Dial/DialSession.cs
index adee1d9..af704a6 100644
--- a/backend/agent/src/MoShiro.Agent/Dial/DialSession.cs
+++ b/backend/agent/src/MoShiro.Agent/Dial/DialSession.cs
@@ -55,7 +55,7 @@ public sealed class DialSession
}
}
- private static bool IsTcpKind(string kind) => kind is "tcp_echo" or "connect" or "hold" or "tcp_bulk";
+ private static bool IsTcpKind(string kind) => kind is "tcp_ping" or "tcp_connect" or "tls_hold" or "tcp_transfer";
/// 调和线路任务：参数没有变化的积木继续运行，其余启动或停止。返回被拒绝的任务（不认识的 Kind、参数非法、超过上限）。
public async Task> ApplyAsync(LineTaskSet set)
@@ -137,7 +137,7 @@ public sealed class DialSession
return false;
}
- if (start.Kind == "udp_train")
+ if (start.Kind == "quic_stream")
{
double mbps = start.Params["mbps"].GetDouble();
long size = start.Params["size"].GetInt64();
@@ -167,7 +167,7 @@ public sealed class DialSession
}
// 最后一批包（单向数据流）、最后几个回显的回复还在途中
- await Task.Delay(TimeSpan.FromMilliseconds(phase.Kind == "udp_train" ? 200 : 300), BlockEnv.Time, ct);
+ await Task.Delay(TimeSpan.FromMilliseconds(phase.Kind == "quic_stream" ? 200 : 300), BlockEnv.Time, ct);
await phase.Running.StopAsync();
return phase.Running.Block.Flush(FlushMode.All, end.SentCount);
}
diff --git a/backend/agent/src/MoShiro.Agent/Tasks/TargetRunners.cs b/backend/agent/src/MoShiro.Agent/Tasks/TargetRunners.cs
index 45a8b17..6b14ded 100644
--- a/backend/agent/src/MoShiro.Agent/Tasks/TargetRunners.cs
+++ b/backend/agent/src/MoShiro.Agent/Tasks/TargetRunners.cs
@@ -7,14 +7,16 @@ using MoShiro.Protocol;
namespace MoShiro.Agent;
///
-/// 目标任务（Agent 自己探测一个地址，没有对端）：用和线路任务同一套积木（ 里的拨号方，例如 icmp_echo、connect），
-/// 只是列块不交给 Agent 合并，而是自己盖章、10 秒一块直接上传。目标地址必须在本地 allowed_target_cidrs 内
-/// （由 在接受任务之前检查）。
+/// 目标任务（Agent 自己探测一个地址，没有对端）：用和线路任务同一套积木（ 里的拨号方，例如 icmp_ping、connect），
+/// 只是列块不交给 Agent 合并，而是自己盖章、10 秒一块直接上传。目标可以是 IP、域名或（http）URL；域名在 Agent 上解析，
+/// 每个解析出的地址都必须在本地 allowed_target_cidrs 内（IP 目标由 在接受任务之前检查，域名在这里检查）。
///
public sealed class TargetRunners
{
private static readonly TimeSpan s_chunkSpan = TimeSpan.FromSeconds(ProtocolLimits.ChunkSeconds);
+ private static readonly TimeSpan s_resolveRetry = TimeSpan.FromMinutes(1);
+
private readonly Dictionary _running = [];
private readonly object _gate = new();
private readonly ChunkAssembler _chunks;
@@ -33,7 +35,7 @@ public sealed class TargetRunners
}
/// 让运行中的任务与期望集合一致：参数没有变化的保持，其余启动或停止。
- public void Reconcile(IReadOnlyList<(TaskSpec Spec, TaskDecision Decision)> desired)
+ public void Reconcile(IReadOnlyList<(TaskSpec Spec, TaskDecision Decision)> desired, Func? allow = null)
{
lock (_gate)
{
@@ -56,7 +58,7 @@ public sealed class TargetRunners
var cts = new CancellationTokenSource();
_running[id] = new Running(Signature(spec), cts);
- _ = Task.Run(() => RunAsync(spec, decision, cts.Token), CancellationToken.None);
+ _ = Task.Run(() => RunAsync(spec, decision, allow, cts.Token), CancellationToken.None);
}
}
}
@@ -64,23 +66,33 @@ public sealed class TargetRunners
private static string Signature(TaskSpec spec) =>
spec.Kind + "|" + spec.Target + "|" + string.Join(',', spec.Params.OrderBy(p => p.Key, StringComparer.Ordinal).Select(p => p.Key + "=" + p.Value.GetRawText()));
- private async Task RunAsync(TaskSpec spec, TaskDecision decision, CancellationToken ct)
+ private async Task RunAsync(TaskSpec spec, TaskDecision decision, Func? allow, CancellationToken ct)
{
try
{
- if (!TryResolve(spec.Target!, decision.Resolved!, spec.Kind, out IPAddress? address, out int port))
+ if (!TargetHost.TryParse(spec.Target, out string host, out int targetPort, out _))
{
- _log.Warn("target is not an address", ("task", spec.Id.Value));
+ _log.Warn("target is not valid", ("task", spec.Id.Value));
return;
}
+ // http 自己解析并检查每个地址；其余 Kind 在这里解析出一个允许的地址作为对端
+ IPAddress? address = null;
+ if (spec.Kind != TargetHost.HttpKind)
+ {
+ address = await ResolveAllowedAsync(spec, host, allow, ct);
+ }
+
+ int port = targetPort != 0 ? targetPort : spec.Kind == "tcp_connect" ? (int)decision.Resolved!.GetInt("port") : 0;
var env = new BlockEnv
{
Clock = _clock,
Log = _log,
Session = () => null,
PeerAddress = address,
- Fragments = new LocalSink(this, spec, address.AddressFamily == System.Net.Sockets.AddressFamily.InterNetworkV6 ? IpVersions.V6 : IpVersions.V4),
+ Target = spec.Target,
+ AllowAddress = allow,
+ Fragments = new LocalSink(this, spec, address?.AddressFamily == System.Net.Sockets.AddressFamily.InterNetworkV6 ? IpVersions.V6 : IpVersions.V4),
};
IBlock block = BlockCatalog.Find(spec.Kind)!.Dialer(env, new BlockSpec(decision.Kind!, decision.Resolved!, 0, port, new FragmentScope(spec.Id, null, null)));
using var cts = CancellationTokenSource.CreateLinkedTokenSource(ct);
@@ -110,21 +122,29 @@ public sealed class TargetRunners
}
}
- private static bool TryResolve(string target, ParamValues values, string kind, out IPAddress address, out int port)
+ /// 解析目标，取第一个允许的地址；解析失败或没有允许的地址就每分钟重试（域名的解析结果会变）。
+ private async Task ResolveAllowedAsync(TaskSpec spec, string host, Func? allow, CancellationToken ct)
{
- port = kind == "connect" ? (int)values.GetInt("port") : 0;
- if (IPEndPoint.TryParse(target, out IPEndPoint? endpoint))
+ while (true)
{
- address = endpoint.Address;
- if (endpoint.Port != 0)
+ try
{
- port = endpoint.Port;
+ IPAddress[] addresses = IPAddress.TryParse(host, out IPAddress? literal) ? [literal] : await Dns.GetHostAddressesAsync(host, ct);
+ IPAddress? chosen = addresses.FirstOrDefault(a => allow?.Invoke(a) ?? true);
+ if (chosen is not null)
+ {
+ return chosen;
+ }
+
+ _log.Warn("no address of the target is allowed", ("task", spec.Id.Value), ("host", host));
+ }
+ catch (System.Net.Sockets.SocketException ex)
+ {
+ _log.Warn("target did not resolve", ("task", spec.Id.Value), ("host", host), ("err", ex.Message));
}
- return true;
+ await Task.Delay(s_resolveRetry, _time, ct);
}
-
- return IPAddress.TryParse(target, out address!);
}
/// 目标任务的行没有拨号方一侧可以合并：积木交出的就是完整的行，Agent 自己盖章（目标、时钟锚点）并放进发件箱。
diff --git a/backend/agent/src/MoShiro.Agent/Tasks/TaskManager.cs b/backend/agent/src/MoShiro.Agent/Tasks/TaskManager.cs
index f2ceaa2..a217b68 100644
--- a/backend/agent/src/MoShiro.Agent/Tasks/TaskManager.cs
+++ b/backend/agent/src/MoShiro.Agent/Tasks/TaskManager.cs
@@ -15,7 +15,7 @@ namespace MoShiro.Agent;
///
public sealed class TaskManager
{
- // 线路任务 = 有监听方积木、并且不是只用于按需测试的 Kind（udp_train、tcp_bulk 会占满链路）；目标任务 = 可以由 Agent 自己运行的 Kind
+ // 线路任务 = 有监听方积木、并且不是只用于按需测试的 Kind（quic_stream、tcp_transfer 会占满链路）；目标任务 = 可以由 Agent 自己运行的 Kind
private static readonly HashSet s_lineKinds = [.. BlockCatalog.All.Where(p => p.Listener is not null && !p.OnDemandOnly).Select(p => p.Kind)];
private static readonly HashSet s_targetKinds = [.. BlockCatalog.All.Where(p => p.RunsAsTarget).Select(p => p.Kind)];
@@ -151,7 +151,7 @@ public sealed class TaskManager
}
}
- _targets.Reconcile(targets);
+ _targets.Reconcile(targets, guard.AddressAllowed);
// 参数没有变化的任务沿用原来的流和积木，数据面上的会话（重放窗口、记录）不被打断
var retired = new List();
diff --git a/backend/agent/src/MoShiro.Agent/Tests/TestRunner.cs b/backend/agent/src/MoShiro.Agent/Tests/TestRunner.cs
index 236d574..7f8ad62 100644
--- a/backend/agent/src/MoShiro.Agent/Tests/TestRunner.cs
+++ b/backend/agent/src/MoShiro.Agent/Tests/TestRunner.cs
@@ -206,7 +206,7 @@ public sealed class TestRunner
}
// 单向数据流的发送端自己运行到结束（或失败）；其他积木是反应式的，等够时长即可
- if (phase.Kind == "udp_train")
+ if (phase.Kind == "quic_stream")
{
try
{
diff --git a/backend/fleet/src/Fleet.Link/FleetLink.cs b/backend/fleet/src/Fleet.Link/FleetLink.cs
index d6c8248..fe9a21d 100644
--- a/backend/fleet/src/Fleet.Link/FleetLink.cs
+++ b/backend/fleet/src/Fleet.Link/FleetLink.cs
@@ -55,7 +55,7 @@ public sealed class FleetLinkOptions
/// 服务端至少要是哪个版本（每次握手重新取）；空表示不要求。不够就关掉这条连接，等 再握手，服务端可能正在更新。
public Func? MinServerVersion { get; init; }
- /// 框架之外的能力，例如 moshiro.v1 与 moshiro.kinds:udp_echo,tcp_echo。
+ /// 框架之外的能力，例如 moshiro.v1 与 moshiro.kinds:quic_ping,tcp_ping。
public IReadOnlyList Capabilities { get; init; } = [];
/// 附加到 fleet.hello 的产品元数据。
diff --git a/backend/shared/MoShiro.Measurement/Blocks/BlockBase.cs b/backend/shared/MoShiro.Measurement/Blocks/BlockBase.cs
index 60e28e0..478b75b 100644
--- a/backend/shared/MoShiro.Measurement/Blocks/BlockBase.cs
+++ b/backend/shared/MoShiro.Measurement/Blocks/BlockBase.cs
@@ -76,7 +76,7 @@ public abstract class BlockBase : IBlock
}
}
- /// 读这个任务的自适应频率参数（udp_echo 和 tcp_echo 共用）。
+ /// 读这个任务的自适应频率参数（quic_ping 和 tcp_ping 共用）。
protected AdaptiveRate? CreateAdaptiveRate()
{
if (!Spec.Bool("adaptive"))
diff --git a/backend/shared/MoShiro.Measurement/Blocks/BlockCatalog.cs b/backend/shared/MoShiro.Measurement/Blocks/BlockCatalog.cs
index 3e4bf44..665bfd1 100644
--- a/backend/shared/MoShiro.Measurement/Blocks/BlockCatalog.cs
+++ b/backend/shared/MoShiro.Measurement/Blocks/BlockCatalog.cs
@@ -26,13 +26,14 @@ public static class BlockCatalog
{
public static readonly IReadOnlyList All =
[
- new("udp_echo", (e, s) => new DatagramEchoDialer(e, s), (e, s) => new DatagramEchoListener(e, s)),
- new("tcp_echo", (e, s) => new TcpEchoDialer(e, s), (e, s) => new TcpEchoListener(e, s)),
- new("udp_train", (e, s) => new TrainDialer(e, s), (e, s) => new TrainListener(e, s), OnDemandOnly: true, Directions: ["rev"]),
- new("tcp_bulk", (e, s) => new TcpBulkDialer(e, s), (e, s) => new TcpBulkListener(e, s), OnDemandOnly: true, Directions: ["fwd", "rev"]),
- new("connect", (e, s) => new ConnectDialer(e, s), (e, s) => new IdleListener(e, s), RunsAsTarget: true),
- new("hold", (e, s) => new HoldDialer(e, s), (e, s) => new TcpEchoListener(e, s, record: false)),
- new("icmp_echo", (e, s) => new IcmpEchoSolo(e, s), null, RunsAsTarget: true),
+ new("quic_ping", (e, s) => new DatagramEchoDialer(e, s), (e, s) => new DatagramEchoListener(e, s)),
+ new("tcp_ping", (e, s) => new TcpEchoDialer(e, s), (e, s) => new TcpEchoListener(e, s)),
+ new("quic_stream", (e, s) => new TrainDialer(e, s), (e, s) => new TrainListener(e, s), OnDemandOnly: true, Directions: ["rev"]),
+ new("tcp_transfer", (e, s) => new TcpBulkDialer(e, s), (e, s) => new TcpBulkListener(e, s), OnDemandOnly: true, Directions: ["fwd", "rev"]),
+ new("tcp_connect", (e, s) => new ConnectDialer(e, s), (e, s) => new IdleListener(e, s), RunsAsTarget: true),
+ new("tls_hold", (e, s) => new HoldDialer(e, s), (e, s) => new TcpEchoListener(e, s, record: false)),
+ new("http_request", (e, s) => new HttpDialer(e, s), null, RunsAsTarget: true),
+ new("icmp_ping", (e, s) => new IcmpEchoSolo(e, s), null, RunsAsTarget: true),
];
public static BlockPair? Find(string kind) => All.FirstOrDefault(p => p.Kind == kind);
diff --git a/backend/shared/MoShiro.Measurement/Blocks/Http.cs b/backend/shared/MoShiro.Measurement/Blocks/Http.cs
new file mode 100644
index 0000000..1a2c630
--- /dev/null
+++ b/backend/shared/MoShiro.Measurement/Blocks/Http.cs
@@ -0,0 +1,382 @@
+using System.Net;
+using System.Net.Http;
+using System.Net.Quic;
+using System.Net.Security;
+using System.Net.Sockets;
+using System.Security.Authentication;
+using Fleet.Core;
+using MoShiro.Protocol;
+
+namespace MoShiro.Measurement;
+
+/// http 请求的结果，编码与 Backend 的 HttpOutcome 一致。
+internal static class HttpOutcomes
+{
+ public const int Ok = 0;
+ public const int Timeout = 1;
+ public const int Reset = 2;
+ public const int Tls = 3;
+ public const int Other = 4;
+ public const int Dns = 5;
+ public const int BadStatus = 6;
+ public const int Unsupported = 7;
+ public const int NotAllowed = 8;
+}
+
+/// 解析出的地址都不在本地允许的网段内。
+internal sealed class TargetNotAllowedException : Exception
+{
+ public TargetNotAllowedException(string host)
+ : base("no address of " + host + " is allowed by allowed_target_cidrs")
+ {
+ }
+}
+
+///
+/// 单个 Agent 去请求一个 URL（HTTP/1.1、HTTP/2 或 HTTP/3，不回退），记录 DNS、TCP 建连、首字节、总耗时和状态码。
+/// 每次请求新建连接，所以每次都包含完整的建连过程；频率按失败率自适应：目标一直应答就降到证明得了成功率的最低频率，
+/// 出现失败升到最高频率，全部失败时指数退避。
+///
+public sealed class HttpDialer : BlockBase
+{
+ private readonly DriverLog _log;
+ private readonly int _attempt;
+ private readonly int _ts;
+ private readonly int _w;
+ private readonly int _dns;
+ private readonly int _tcp;
+ private readonly int _ttfb;
+ private readonly int _total;
+ private readonly int _status;
+ private readonly int _proto;
+ private readonly int _outcome;
+ private readonly HashSet _loggedOutcomes = [];
+ private long _next;
+
+ public HttpDialer(BlockEnv env, BlockSpec spec)
+ : base(env, spec)
+ {
+ _log = new DriverLog(spec.Kind.FindStream("req")!, Side.Dialer, 1024);
+ _attempt = _log.Store.ColumnIndex("attempt");
+ _ts = _log.Store.ColumnIndex("ts");
+ _w = _log.Store.ColumnIndex("w");
+ _dns = _log.Store.ColumnIndex("dns_us");
+ _tcp = _log.Store.ColumnIndex("tcp_us");
+ _ttfb = _log.Store.ColumnIndex("ttfb_us");
+ _total = _log.Store.ColumnIndex("total_us");
+ _status = _log.Store.ColumnIndex("status");
+ _proto = _log.Store.ColumnIndex("proto");
+ _outcome = _log.Store.ColumnIndex("outcome");
+ }
+
+ public override long Produced => _log.Count;
+
+ public override async Task RunAsync(CancellationToken ct)
+ {
+ if (!TargetHost.TryParse(Env.Target, out _, out _, out Uri? url) || url is null)
+ {
+ Env.Log.Warn("http target is not a URL", ("task", Spec.Scope.Task?.Value));
+ await WaitUntilCancelledAsync(ct);
+ return;
+ }
+
+ AdaptiveRate? rate = CreateRequestRate();
+ try
+ {
+ if (rate is not null)
+ {
+ await PaceAsync(rate, async () =>
+ {
+ int weight = rate.Weight;
+ long sent = Env.Clock.NowUs();
+ rate.OnSend(sent);
+ bool ok = await RequestAsync(url, weight, ct);
+ if (ok)
+ {
+ long now = Env.Clock.NowUs();
+ rate.OnReply(sent, now, (now - sent) / 1000.0);
+ }
+ }, ct);
+ }
+ else
+ {
+ TimeSpan interval = TimeSpan.FromSeconds(60.0 / Spec.Int("per_minute"));
+ while (!ct.IsCancellationRequested)
+ {
+ long start = Env.Clock.NowUs();
+ await RequestAsync(url, 1, ct);
+ TimeSpan wait = interval - TimeSpan.FromMicroseconds(Env.Clock.NowUs() - start);
+ if (wait > TimeSpan.Zero)
+ {
+ await Task.Delay(wait, Env.Time, ct);
+ }
+ }
+ }
+ }
+ catch (OperationCanceledException)
+ {
+ // 积木被停止
+ }
+ }
+
+ public override int Flush(FlushMode mode, long peerProduced = 0)
+ {
+ RawStream? stream = _log.Take(null, all: true);
+ if (stream is null)
+ {
+ return 0;
+ }
+
+ Ship(stream, "ts");
+ return 1;
+ }
+
+ ///
+ /// 频率以「次/分钟」声明，换算成 Hz 交给 。请求间隔以秒到分钟计，所以窗口按分钟级放宽，
+ /// 并且不用 RTT 分位数的精度保护（样本太少，估计不出来）：平稳时只由成功率的上界决定频率。
+ ///
+ private AdaptiveRate? CreateRequestRate()
+ {
+ if (!Spec.Bool("adaptive"))
+ {
+ return null;
+ }
+
+ double max = Math.Max(Spec.Int("rate_max_per_min"), Spec.Int("rate_min_per_min")) / 60.0;
+ return new AdaptiveRate(new AdaptiveRateOptions
+ {
+ MinHz = Spec.Int("rate_min_per_min") / 60.0,
+ MaxHz = max,
+ TauPct = Spec.Number("tau_pct"),
+ RttGuard = false,
+ CertifySeconds = 900,
+ QuietCertifySeconds = 1200,
+ GraceSeconds = Math.Ceiling(Spec.Int("timeout_ms") / 1000.0) + 1,
+ DeadWindowSeconds = 120,
+ DeadMinSent = 3,
+ DeadMaxIntervalSeconds = 300,
+ ActiveLossCount = 2,
+ Prior = Env.LossPrior,
+ });
+ }
+
+ /// 发一次请求并记一行；返回这次算不算成功（有应答且状态码合格）。
+ private async Task RequestAsync(Uri url, int weight, CancellationToken ct)
+ {
+ string version = Spec.Text("version");
+ bool verify = Spec.Bool("tls_verify");
+ TimeSpan timeout = TimeSpan.FromMilliseconds(Spec.Int("timeout_ms"));
+ var phases = new Phases();
+ long start = Env.Clock.NowUs();
+ int outcome = HttpOutcomes.Other;
+ int status = 0;
+ int proto = 0;
+ long? ttfb = null;
+ long? total = null;
+ using var timeoutCts = new CancellationTokenSource(timeout, Env.Time);
+ using CancellationTokenSource linked = CancellationTokenSource.CreateLinkedTokenSource(ct, timeoutCts.Token);
+ try
+ {
+ if (version == "h3" && !QuicConnection.IsSupported)
+ {
+ outcome = HttpOutcomes.Unsupported;
+ LogFirstFailure(outcome, url, version, null);
+ }
+ else
+ {
+ if (version == "h3")
+ {
+ await ResolveForQuicAsync(url.IdnHost, phases, linked.Token);
+ }
+
+ using HttpMessageInvoker invoker = CreateInvoker(version, verify, phases);
+ using var request = new HttpRequestMessage(new HttpMethod(Spec.Text("method")), url)
+ {
+ Version = version switch { "h1" => HttpVersion.Version11, "h2" => HttpVersion.Version20, _ => HttpVersion.Version30 },
+ VersionPolicy = HttpVersionPolicy.RequestVersionExact,
+ };
+ using HttpResponseMessage response = await invoker.SendAsync(request, linked.Token);
+ ttfb = Env.Clock.NowUs() - start;
+ status = (int)response.StatusCode;
+ proto = response.Version.Major switch { 3 => 3, 2 => 2, _ => 1 };
+ await ReadBodyAsync(response, (int)Spec.Int("read_kb") * 1024, linked.Token);
+ total = Env.Clock.NowUs() - start;
+ long expected = Spec.Int("ok_status");
+ outcome = (expected == 0 ? status < 400 : status == expected) ? HttpOutcomes.Ok : HttpOutcomes.BadStatus;
+ }
+ }
+ catch (OperationCanceledException) when (!ct.IsCancellationRequested)
+ {
+ outcome = HttpOutcomes.Timeout;
+ }
+ catch (OperationCanceledException)
+ {
+ return false;
+ }
+ catch (Exception ex) when (ex is HttpRequestException or IOException or SocketException or AuthenticationException or TargetNotAllowedException or QuicException)
+ {
+ outcome = Classify(ex);
+ LogFirstFailure(outcome, url, version, ex);
+ }
+
+ Record(start, weight, phases, ttfb, total, status, proto, outcome);
+ return outcome == HttpOutcomes.Ok;
+ }
+
+ private HttpMessageInvoker CreateInvoker(string version, bool verify, Phases phases)
+ {
+ var handler = new SocketsHttpHandler
+ {
+ UseProxy = false,
+ AllowAutoRedirect = false,
+ UseCookies = false,
+ AutomaticDecompression = DecompressionMethods.None,
+ };
+
+ // 设置了 ConnectCallback 的处理器不能用 HTTP/3（运行时用自己的 QUIC 连接，不经过回调）：h3 只能靠请求前的解析和网段检查
+ if (version != "h3")
+ {
+ handler.ConnectCallback = (context, token) => ConnectAsync(context.DnsEndPoint, phases, token);
+ }
+
+ if (!verify)
+ {
+#pragma warning disable CA5359 // 操作员明确关闭证书校验（tls_verify=false）：只测可达性和耗时
+ handler.SslOptions = new SslClientAuthenticationOptions { RemoteCertificateValidationCallback = (_, _, _, _) => true };
+#pragma warning restore CA5359
+ }
+
+ return new HttpMessageInvoker(handler, disposeHandler: true);
+ }
+
+ /// 自己解析和连接：每个地址都要在允许的网段里，DNS 和 TCP 建连的耗时分别记下来。
+ private async ValueTask ConnectAsync(DnsEndPoint endpoint, Phases phases, CancellationToken ct)
+ {
+ IPAddress[] addresses = await ResolveAsync(endpoint.Host, phases, ct);
+ long start = Env.Clock.NowUs();
+ Exception? last = null;
+ foreach (IPAddress address in addresses)
+ {
+ var socket = new Socket(address.AddressFamily, SocketType.Stream, ProtocolType.Tcp) { NoDelay = true };
+ try
+ {
+ await socket.ConnectAsync(new IPEndPoint(address, endpoint.Port), ct);
+ phases.TcpUs = Env.Clock.NowUs() - start;
+ return new NetworkStream(socket, ownsSocket: true);
+ }
+ catch (SocketException ex)
+ {
+ socket.Dispose();
+ last = ex;
+ }
+ }
+
+ throw last ?? new SocketException((int)SocketError.HostNotFound);
+ }
+
+ /// HTTP/3 由运行时自己建 UDP 连接，没法接管；先解析并检查一遍，再让它去连。
+ private async Task ResolveForQuicAsync(string host, Phases phases, CancellationToken ct) => _ = await ResolveAsync(host, phases, ct);
+
+ private async Task ResolveAsync(string host, Phases phases, CancellationToken ct)
+ {
+ long start = Env.Clock.NowUs();
+ IPAddress[] all = IPAddress.TryParse(host, out IPAddress? literal) ? [literal] : await Dns.GetHostAddressesAsync(host, ct);
+ phases.DnsUs = Env.Clock.NowUs() - start;
+ IPAddress[] allowed = [.. all.Where(a => Env.AllowAddress?.Invoke(a) ?? true)];
+ return allowed.Length > 0 ? allowed : throw new TargetNotAllowedException(host);
+ }
+
+ private static async Task ReadBodyAsync(HttpResponseMessage response, int limit, CancellationToken ct)
+ {
+ if (limit <= 0)
+ {
+ return;
+ }
+
+ await using Stream body = await response.Content.ReadAsStreamAsync(ct);
+ byte[] buffer = new byte[Math.Min(limit, 16 * 1024)];
+ int read = 0;
+ while (read < limit)
+ {
+ int n = await body.ReadAsync(buffer.AsMemory(0, Math.Min(buffer.Length, limit - read)), ct);
+ if (n == 0)
+ {
+ break;
+ }
+
+ read += n;
+ }
+ }
+
+ /// 每种失败结果只记第一次（带原因），保留给事后审计，不会随请求频率刷屏。
+ private void LogFirstFailure(int outcome, Uri url, string version, Exception? ex)
+ {
+ if (!_loggedOutcomes.Add(outcome))
+ {
+ return;
+ }
+
+ Env.Log.Warn("http request failed", ("task", Spec.Scope.Task?.Value), ("outcome", outcome), ("version", version), ("host", url.Host),
+ ("quic_supported", version == "h3" ? QuicConnection.IsSupported : null), ("err", ex is null ? "System.Net.Quic is not available" : ex.GetType().Name + ": " + (ex.InnerException?.Message ?? ex.Message)));
+ }
+
+ private static int Classify(Exception ex)
+ {
+ for (Exception? e = ex; e is not null; e = e.InnerException)
+ {
+ switch (e)
+ {
+ case TargetNotAllowedException:
+ return HttpOutcomes.NotAllowed;
+ case AuthenticationException:
+ return HttpOutcomes.Tls;
+ case SocketException { SocketErrorCode: SocketError.HostNotFound or SocketError.TryAgain or SocketError.NoData }:
+ return HttpOutcomes.Dns;
+ case SocketException { SocketErrorCode: SocketError.ConnectionRefused or SocketError.ConnectionReset }:
+ return HttpOutcomes.Reset;
+ case HttpRequestException { HttpRequestError: HttpRequestError.NameResolutionError }:
+ return HttpOutcomes.Dns;
+ case HttpRequestException { HttpRequestError: HttpRequestError.SecureConnectionError }:
+ return HttpOutcomes.Tls;
+ case HttpRequestException { HttpRequestError: HttpRequestError.VersionNegotiationError }:
+ return HttpOutcomes.Unsupported;
+ }
+ }
+
+ return HttpOutcomes.Other;
+ }
+
+ private void Record(long startUs, int weight, Phases phases, long? ttfb, long? total, int status, int proto, int outcome)
+ {
+ long key = _log.Store.Append();
+ _log.Store.Set(key, _attempt, _next++);
+ _log.Store.Set(key, _ts, startUs);
+ _log.Store.Set(key, _w, Math.Clamp(weight, 1, 65535));
+ _log.Store.Set(key, _outcome, outcome);
+ SetIf(key, _dns, phases.DnsUs);
+ SetIf(key, _tcp, phases.TcpUs);
+ SetIf(key, _ttfb, ttfb);
+ SetIf(key, _total, total);
+ if (status != 0)
+ {
+ _log.Store.Set(key, _status, status);
+ _log.Store.Set(key, _proto, proto);
+ }
+ }
+
+ private void SetIf(long key, int column, long? value)
+ {
+ if (value is { } v)
+ {
+ _log.Store.Set(key, column, Math.Clamp(v, 0, uint.MaxValue));
+ }
+ }
+
+ /// 一次请求里 DNS 和 TCP 建连用了多久（连接回调里填）。
+ private sealed class Phases
+ {
+ public long? DnsUs { get; set; }
+
+ public long? TcpUs { get; set; }
+ }
+}
diff --git a/backend/shared/MoShiro.Measurement/Blocks/Train.cs b/backend/shared/MoShiro.Measurement/Blocks/Train.cs
index cafd7da..2bca408 100644
--- a/backend/shared/MoShiro.Measurement/Blocks/Train.cs
+++ b/backend/shared/MoShiro.Measurement/Blocks/Train.cs
@@ -219,7 +219,7 @@ public sealed class TrainDialer : BlockBase
[FragmentKeys.SeqTo] = Number(to),
};
Env.Fragments!.Ship(
- "udp_train",
+ "quic_stream",
Spec.Scope,
fragments == 0 ? [pkt, _step.Cut(0, 1, includeKey: false)] : [pkt],
parameters,
diff --git a/backend/shared/MoShiro.Measurement/Core/AdaptiveRate.cs b/backend/shared/MoShiro.Measurement/Core/AdaptiveRate.cs
index b7eb0bc..19dd130 100644
--- a/backend/shared/MoShiro.Measurement/Core/AdaptiveRate.cs
+++ b/backend/shared/MoShiro.Measurement/Core/AdaptiveRate.cs
@@ -65,6 +65,9 @@ public sealed record AdaptiveRateOptions
public double RttFloorMs { get; init; } = 5;
+ /// 用 RTT p99 的置信区间限制降频。样本稀少的任务（每分钟几次）估不出区间，关掉后平稳频率只由丢包上界决定。
+ public bool RttGuard { get; init; } = true;
+
/// 先验：时间 → 这个时刻丢包的历史概率（0 到 1）；null 表示没有历史。
public Func? Prior { get; init; }
@@ -285,7 +288,7 @@ public sealed class AdaptiveRate
UpdateGuard(nowUs);
double hz = next switch
{
- ProbeState.Quiet => Quantize(Math.Max(QuietHz(QuietWindowSeconds(nowUs)), double.IsNaN(_guardHz) ? _options.MaxHz : _guardHz)),
+ ProbeState.Quiet => Quantize(Math.Max(QuietHz(QuietWindowSeconds(nowUs)), !_options.RttGuard ? _options.MinHz : double.IsNaN(_guardHz) ? _options.MaxHz : _guardHz)),
ProbeState.Watch or ProbeState.Active => _options.MaxHz,
_ => Math.Max(_options.MinHz / 4, 1.0 / _options.DeadMaxIntervalSeconds),
};
diff --git a/backend/shared/MoShiro.Measurement/Core/Contracts.cs b/backend/shared/MoShiro.Measurement/Core/Contracts.cs
index dd1fd9d..44519c5 100644
--- a/backend/shared/MoShiro.Measurement/Core/Contracts.cs
+++ b/backend/shared/MoShiro.Measurement/Core/Contracts.cs
@@ -190,6 +190,8 @@ public sealed class BlockSpec
public bool Bool(string name) => Params.GetBool(name);
+ public string Text(string name) => Params.GetString(name);
+
/// 参数或端口变化时调和器重启这个积木（不含流编号：Agent 的流编号每次重建都会重新分配）。
public string ConfigSignature => KindId + "|" + Port + "|" +
string.Join(',', Params.Values.OrderBy(p => p.Key, StringComparer.Ordinal).Select(p => p.Key + "=" + p.Value.GetRawText()));
@@ -237,6 +239,12 @@ public sealed class BlockEnv
/// 拨号方：对端（Agent，或目标任务的目标地址）。
public IPAddress? PeerAddress { get; init; }
+ /// 单个 Agent 任务的目标原文（http 是 URL，其余是主机[:端口]）；线路任务为 null。
+ public string? Target { get; init; }
+
+ /// 单个 Agent 任务：解析出来的地址能不能连（本地 allowed_target_cidrs）；null 表示不限制。
+ public Func? AllowAddress { get; init; }
+
/// 拨号方：对端 TLS 证书的 SPKI 指纹；null 表示不校验证书（目标任务，目标是任意地址）。
public byte[]? PeerSpkiSha256 { get; init; }
@@ -246,7 +254,7 @@ public sealed class BlockEnv
/// 拨号方：列块出口。
public IFragmentSink? Fragments { get; init; }
- /// 本地硬上限：批量传输（tcp_bulk）发送端的码率（Mbit/s）。批量传输会尽可能占满链路，这是防止把小机器打满的保险。
+ /// 本地硬上限：批量传输（tcp_transfer）发送端的码率（Mbit/s）。批量传输会尽可能占满链路，这是防止把小机器打满的保险。
public double MaxBulkMbps { get; init; } = 100;
/// 本地硬上限：单向数据流接收缓冲的包速率上界。
@@ -267,6 +275,8 @@ public sealed class BlockEnv
Counters = Counters,
TcpListen = TcpListen,
PeerAddress = PeerAddress,
+ Target = Target,
+ AllowAddress = AllowAddress,
PeerSpkiSha256 = PeerSpkiSha256,
KnockKey = KnockKey,
Fragments = Fragments,
diff --git a/backend/shared/MoShiro.Measurement/Execution/BudgetGuard.cs b/backend/shared/MoShiro.Measurement/Execution/BudgetGuard.cs
index 06bd324..f589594 100644
--- a/backend/shared/MoShiro.Measurement/Execution/BudgetGuard.cs
+++ b/backend/shared/MoShiro.Measurement/Execution/BudgetGuard.cs
@@ -106,22 +106,25 @@ public sealed class BudgetGuard
return Check(new TaskSpec { Id = new TaskId("tsk_phase"), Scope = "line", Kind = kindId, Params = new Dictionary(parameters) }, ref mbps, ref pps);
}
+ ///
+ /// 目标在静态上可不可以接受：IP 字面量按 allowed_target_cidrs 检查；域名（包括 URL 里的）要解析之后才知道，
+ /// 这里只要求配置过白名单，运行时每次解析出的地址再用 检查。
+ ///
public bool TargetAllowed(string? target)
{
- if (target is null)
+ if (!TargetHost.TryParse(target, out string? host, out _, out _))
{
return false;
}
- string host = target;
- if (IPEndPoint.TryParse(target, out IPEndPoint? endpoint))
- {
- host = endpoint.Address.ToString();
- }
+ return IPAddress.TryParse(host, out IPAddress? address) ? AddressAllowed(address) : _cidrs.Count > 0;
+ }
- if (!IPAddress.TryParse(host, out IPAddress? address))
+ public bool AddressAllowed(IPAddress address)
+ {
+ if (address.IsIPv4MappedToIPv6)
{
- return false;
+ address = address.MapToIPv4();
}
foreach ((IPAddress network, int prefix) in _cidrs)
@@ -148,10 +151,11 @@ public sealed class BudgetGuard
private static long EstimatePps(string kindId, ParamValues values) => kindId switch
{
- "udp_echo" or "tcp_echo" => Kinds.PeakRateHz(kindId, values) * 2,
- "icmp_echo" => values.GetInt("rate_hz") * 2,
- "udp_train" => (long)(values.GetNumber("mbps") * 1_000_000 / 8 / Math.Max(1, values.GetInt("size"))),
- "connect" => values.GetInt("per_minute") * 6 / 60,
+ "quic_ping" or "tcp_ping" => Kinds.PeakRateHz(kindId, values) * 2,
+ "icmp_ping" => values.GetInt("rate_hz") * 2,
+ "quic_stream" => (long)(values.GetNumber("mbps") * 1_000_000 / 8 / Math.Max(1, values.GetInt("size"))),
+ "tcp_connect" => values.GetInt("per_minute") * 6 / 60,
+ "http_request" => Math.Max(1, values.GetInt("rate_max_per_min") * 20 / 60),
_ => 1,
};
diff --git a/backend/shared/MoShiro.Measurement/Net/MsQuicApi.cs b/backend/shared/MoShiro.Measurement/Net/MsQuicApi.cs
index e997628..2bddd21 100644
--- a/backend/shared/MoShiro.Measurement/Net/MsQuicApi.cs
+++ b/backend/shared/MoShiro.Measurement/Net/MsQuicApi.cs
@@ -109,14 +109,16 @@ internal static unsafe class MsQuicApi
public static void RegistrationClose(nint handle) => s_registrationClose(handle);
- public static nint ConfigurationOpen(nint registration, byte[] alpn)
+ public static nint ConfigurationOpen(nint registration, byte[] alpn, QuicSettings? settings = null)
{
+ byte[] native = settings is { IsDefault: false } ? settings.ToNative() : [];
unsafe
{
fixed (byte* alpnBytes = alpn)
+ fixed (byte* settingsBytes = native)
{
var buffer = new QuicBuffer { Length = (uint)alpn.Length, Buffer = (nint)alpnBytes };
- uint status = s_configurationOpen(registration, ref buffer, 1, 0, 0, 0, out nint configuration);
+ uint status = s_configurationOpen(registration, ref buffer, 1, native.Length == 0 ? 0 : (nint)settingsBytes, (uint)native.Length, 0, out nint configuration);
Check(status, "ConfigurationOpen");
return configuration;
}
diff --git a/backend/shared/MoShiro.Measurement/Net/MsQuicConnection.cs b/backend/shared/MoShiro.Measurement/Net/MsQuicConnection.cs
index 45ef668..db9bbe6 100644
--- a/backend/shared/MoShiro.Measurement/Net/MsQuicConnection.cs
+++ b/backend/shared/MoShiro.Measurement/Net/MsQuicConnection.cs
@@ -156,7 +156,8 @@ internal sealed class MsQuicConnection : IAsyncDisposable, IDatagramConnection
byte[] expectedPeerSpki,
MsQuicDatagramCallback datagramHandler,
INodeLog log,
- CancellationToken ct)
+ CancellationToken ct,
+ QuicSettings? settings = null)
{
nint registration = 0;
nint configuration = 0;
@@ -165,7 +166,7 @@ internal sealed class MsQuicConnection : IAsyncDisposable, IDatagramConnection
try
{
registration = MsQuicApi.RegistrationOpen("moshiro-dialer");
- configuration = MsQuicApi.ConfigurationOpen(registration, QuicAlpn.Value);
+ configuration = MsQuicApi.ConfigurationOpen(registration, QuicAlpn.Value, settings);
LoadClientCredentials(configuration);
var endpoint = new IPEndPoint(address.AddressFamily == AddressFamily.InterNetworkV6 ? IPAddress.IPv6Any : IPAddress.Any, localPort);
SocketAddress serialized = endpoint.Serialize();
diff --git a/backend/shared/MoShiro.Measurement/Net/MsQuicListener.cs b/backend/shared/MoShiro.Measurement/Net/MsQuicListener.cs
index 6201f29..3922b45 100644
--- a/backend/shared/MoShiro.Measurement/Net/MsQuicListener.cs
+++ b/backend/shared/MoShiro.Measurement/Net/MsQuicListener.cs
@@ -21,6 +21,7 @@ internal sealed class MsQuicListener : IAsyncDisposable
private readonly INodeLog _log;
private readonly Action? _peerAddressChanged;
private readonly X509Certificate2 _certificate;
+ private readonly QuicSettings? _settings;
private readonly GCHandle _self;
private readonly ConcurrentDictionary _connections = new();
private readonly List _listeners = [];
@@ -34,8 +35,10 @@ internal sealed class MsQuicListener : IAsyncDisposable
X509Certificate2 certificate,
MsQuicDatagramCallback datagramHandler,
INodeLog log,
- Action? peerAddressChanged)
+ Action? peerAddressChanged,
+ QuicSettings? settings = null)
{
+ _settings = settings;
// MsQuic opens dual-mode wildcard sockets, but filters incoming connections using
// the address family supplied to ListenerStart. An IPv6Any listener alone rejects
// IPv4-mapped connections with TLS alert 120 (no_application_protocol).
@@ -62,7 +65,7 @@ internal sealed class MsQuicListener : IAsyncDisposable
_registration = MsQuicApi.RegistrationOpen("moshiro-agent");
try
{
- _configuration = MsQuicApi.ConfigurationOpen(_registration, QuicAlpn.Value);
+ _configuration = MsQuicApi.ConfigurationOpen(_registration, QuicAlpn.Value, _settings);
MsQuicConnection.LoadServerCredentials(_configuration, _certificate);
var bound = new List(_ports.Count * _bindAddresses.Count);
foreach (int port in _ports)
diff --git a/backend/shared/MoShiro.Measurement/Net/QuicDatagramTransport.cs b/backend/shared/MoShiro.Measurement/Net/QuicDatagramTransport.cs
index 66910dd..aabaf16 100644
--- a/backend/shared/MoShiro.Measurement/Net/QuicDatagramTransport.cs
+++ b/backend/shared/MoShiro.Measurement/Net/QuicDatagramTransport.cs
@@ -21,6 +21,7 @@ public sealed class QuicDatagramTransport : IDisposable, IAsyncDisposable
private readonly DataSession? _clientSession;
private readonly IPEndPoint? _remote;
private readonly byte[]? _peerSpki;
+ private readonly QuicSettings? _settings;
private int _previousSourcePort;
private readonly TaskCompletionSource _clientReady = new(TaskCreationOptions.RunContinuationsAsynchronously);
private MsQuicListener? _listener;
@@ -28,8 +29,9 @@ public sealed class QuicDatagramTransport : IDisposable, IAsyncDisposable
private int _disposed;
/// Constructs an Agent-side QUIC listener.
- public QuicDatagramTransport(IDataSessions sessions, IAdmission? admission, MonoClock clock, DataPlaneCounters counters, INodeLog log, IPAddress bind, IReadOnlyList ports, X509Certificate2 serverCertificate)
+ public QuicDatagramTransport(IDataSessions sessions, IAdmission? admission, MonoClock clock, DataPlaneCounters counters, INodeLog log, IPAddress bind, IReadOnlyList ports, X509Certificate2 serverCertificate, QuicSettings? settings = null)
{
+ _settings = settings;
_router = new MeasurementDatagramRouter(sessions, admission, clock, counters, log);
_log = log;
_onDatagram = _router.Dispatch;
@@ -39,8 +41,9 @@ public sealed class QuicDatagramTransport : IDisposable, IAsyncDisposable
}
/// Constructs a 拨号方-side QUIC dialer for one Agent session.
- public QuicDatagramTransport(DataSession session, MonoClock clock, DataPlaneCounters counters, INodeLog log, IPEndPoint remote, byte[] peerSpki)
+ public QuicDatagramTransport(DataSession session, MonoClock clock, DataPlaneCounters counters, INodeLog log, IPEndPoint remote, byte[] peerSpki, QuicSettings? settings = null)
{
+ _settings = settings;
_router = new MeasurementDatagramRouter(new SingleSession(session), null, clock, counters, log);
_log = log;
_onDatagram = _router.Dispatch;
@@ -73,7 +76,7 @@ public sealed class QuicDatagramTransport : IDisposable, IAsyncDisposable
throw new InvalidOperationException("QUIC listener is already bound.");
}
- _listener = new MsQuicListener(_bind, _ports, _serverCertificate!, _onDatagram, _log, null);
+ _listener = new MsQuicListener(_bind, _ports, _serverCertificate!, _onDatagram, _log, null, _settings);
_listener.Start();
BoundPorts = _listener.BoundPorts;
}
@@ -92,7 +95,7 @@ public sealed class QuicDatagramTransport : IDisposable, IAsyncDisposable
ObjectDisposedException.ThrowIf(Volatile.Read(ref _disposed) != 0, this);
await CloseClientAsync().ConfigureAwait(false);
int sourcePort = SelectSourcePort(_remote.AddressFamily, _previousSourcePort);
- MsQuicConnection client = await MsQuicConnection.ConnectAsync(_remote.Address, _remote.Port, sourcePort, _peerSpki, _onDatagram, _log, ct).ConfigureAwait(false);
+ MsQuicConnection client = await MsQuicConnection.ConnectAsync(_remote.Address, _remote.Port, sourcePort, _peerSpki, _onDatagram, _log, ct, _settings).ConfigureAwait(false);
_client = client;
_clientSession.BindDatagramConnection(client);
BoundPorts = client.LocalEndPoint is { } local ? [local.Port] : [sourcePort];
diff --git a/backend/shared/MoShiro.Measurement/Net/QuicSettings.cs b/backend/shared/MoShiro.Measurement/Net/QuicSettings.cs
new file mode 100644
index 0000000..6eee9a8
--- /dev/null
+++ b/backend/shared/MoShiro.Measurement/Net/QuicSettings.cs
@@ -0,0 +1,110 @@
+using System.Buffers.Binary;
+
+namespace MoShiro.Measurement;
+
+///
+/// 数据通道 QUIC 连接的传输参数（agent.json 的 quic 段）；null 的项用 MsQuic 默认值。
+/// 一条 QUIC 连接由这个 Agent 与对端之间的所有任务共用，所以这些参数属于节点，不属于某个任务。
+/// 监听方用它接受连接，拨号方用它发起连接，两边各自生效。
+///
+public sealed record QuicSettings
+{
+ /// MsQuic 默认的最小、最大 MTU 是 1248 和 1500（IP 包大小，含 IP 和 UDP 头）。
+ public const int MinAllowedMtu = 1200;
+
+ public const int MaxAllowedMtu = 65535;
+
+ /// 路径 MTU 探测的起点；链路的真实 MTU 更小时探测会往下收缩，更大时往上探测到 。
+ public int? MinMtu { get; init; }
+
+ /// 路径 MTU 探测的上限。设成链路的真实 MTU 可以省掉探测，设小可以避开会丢大包的路径。
+ public int? MaxMtu { get; init; }
+
+ public bool? Pacing { get; init; }
+
+ public int? IdleTimeoutMs { get; init; }
+
+ public int? KeepAliveMs { get; init; }
+
+ public bool IsDefault => MinMtu is null && MaxMtu is null && Pacing is null && IdleTimeoutMs is null && KeepAliveMs is null;
+
+ /// 参数是否合法；不合法返回原因。
+ public string? Validate()
+ {
+ if (MinMtu is { } min && min is < MinAllowedMtu or > MaxAllowedMtu)
+ {
+ return $"min_mtu must be {MinAllowedMtu}..{MaxAllowedMtu}";
+ }
+
+ if (MaxMtu is { } max && max is < MinAllowedMtu or > MaxAllowedMtu)
+ {
+ return $"max_mtu must be {MinAllowedMtu}..{MaxAllowedMtu}";
+ }
+
+ if (MinMtu is { } lo && MaxMtu is { } hi && lo > hi)
+ {
+ return "min_mtu must not exceed max_mtu";
+ }
+
+ if (IdleTimeoutMs is < 0)
+ {
+ return "idle_timeout_ms must not be negative";
+ }
+
+ return KeepAliveMs is < 0 ? "keep_alive_ms must not be negative" : null;
+ }
+
+ // QUIC_SETTINGS（msquic.h）的布局：IsSetFlags(u64) 后面依次是 u64、u32、u16 各一组字段，再是位域字节。
+ // 这里只写用到的几项，其余字段保持 0 且不置位。长度只算到 MtuDiscoveryMissingProbeCount，更靠后的字段 MsQuic 不会读。
+ private const int Size = 112;
+ private const int IdleTimeoutOffset = 24;
+ private const int KeepAliveOffset = 88;
+ private const int MinMtuOffset = 102;
+ private const int MaxMtuOffset = 104;
+ private const int BitFieldOffset = 106;
+ private const int PacingBit = 1 << 1;
+
+ private const int IdleTimeoutFlag = 2;
+ private const int KeepAliveFlag = 16;
+ private const int MinMtuFlag = 22;
+ private const int MaxMtuFlag = 23;
+ private const int PacingFlag = 25;
+
+ internal byte[] ToNative()
+ {
+ byte[] native = new byte[Size];
+ ulong isSet = 0;
+ if (IdleTimeoutMs is { } idle)
+ {
+ BinaryPrimitives.WriteUInt64LittleEndian(native.AsSpan(IdleTimeoutOffset), (ulong)idle);
+ isSet |= 1UL << IdleTimeoutFlag;
+ }
+
+ if (KeepAliveMs is { } keepAlive)
+ {
+ BinaryPrimitives.WriteUInt32LittleEndian(native.AsSpan(KeepAliveOffset), (uint)keepAlive);
+ isSet |= 1UL << KeepAliveFlag;
+ }
+
+ if (MinMtu is { } min)
+ {
+ BinaryPrimitives.WriteUInt16LittleEndian(native.AsSpan(MinMtuOffset), (ushort)min);
+ isSet |= 1UL << MinMtuFlag;
+ }
+
+ if (MaxMtu is { } max)
+ {
+ BinaryPrimitives.WriteUInt16LittleEndian(native.AsSpan(MaxMtuOffset), (ushort)max);
+ isSet |= 1UL << MaxMtuFlag;
+ }
+
+ if (Pacing is { } pacing)
+ {
+ native[BitFieldOffset] = (byte)(pacing ? PacingBit : 0);
+ isSet |= 1UL << PacingFlag;
+ }
+
+ BinaryPrimitives.WriteUInt64LittleEndian(native, isSet);
+ return native;
+ }
+}
diff --git a/backend/shared/MoShiro.Protocol/Catalog/Kinds/Kinds.cs b/backend/shared/MoShiro.Protocol/Catalog/Kinds/Kinds.cs
index f20d420..4e4288c 100644
--- a/backend/shared/MoShiro.Protocol/Catalog/Kinds/Kinds.cs
+++ b/backend/shared/MoShiro.Protocol/Catalog/Kinds/Kinds.cs
@@ -10,7 +10,7 @@ public static class Kinds
/// 这个任务可能达到的最高探测频率（Hz）：自适应时是 rate_max_hz，否则是固定的 rate_hz。预算、限速和准入都按它算。
public static long PeakRateHz(string kindId, ParamValues values) =>
- (kindId is "udp_echo" or "tcp_echo") && values.GetBool("adaptive") ? Math.Max(values.GetInt("rate_max_hz"), values.GetInt("rate_min_hz")) : values.GetInt("rate_hz");
+ (kindId is "quic_ping" or "tcp_ping") && values.GetBool("adaptive") ? Math.Max(values.GetInt("rate_max_hz"), values.GetInt("rate_min_hz")) : values.GetInt("rate_hz");
private static readonly StreamSpec s_echoMsg = new(
"msg",
@@ -30,8 +30,8 @@ public static class Kinds
],
Union: true);
- public static readonly KindSpec UdpEcho = new(
- "udp_echo",
+ public static readonly KindSpec QuicPing = new(
+ "quic_ping",
"QUIC Datagram echo",
"Fixed-rate QUIC Datagram echo with four timestamps per packet.",
Scope.Line,
@@ -47,10 +47,10 @@ public static class Kinds
P.Int("port", 7443, 1, 65535),
],
[s_echoMsg],
- p => Bandwidth.UdpEcho(PeakRateHz("udp_echo", p), p.GetInt("size")));
+ p => Bandwidth.UdpEcho(PeakRateHz("quic_ping", p), p.GetInt("size")));
- public static readonly KindSpec UdpTrain = new(
- "udp_train",
+ public static readonly KindSpec QuicStream = new(
+ "quic_stream",
"QUIC Datagram train",
"Fixed-rate or laddered one-way QUIC Datagram stream (Agent to 拨号方) for loss and delay variation under load.",
Scope.Line,
@@ -68,8 +68,8 @@ public static class Kinds
],
p => (long)(p.GetNumber("mbps") * 1_000_000));
- public static readonly KindSpec TcpEcho = new(
- "tcp_echo",
+ public static readonly KindSpec TcpPing = new(
+ "tcp_ping",
"TCP echo",
"Small TCP messages echoed with four timestamps, plus periodic kernel tcp_info samples.",
Scope.Line,
@@ -88,10 +88,10 @@ public static class Kinds
P.Bool("tls", false),
],
[s_echoMsg, s_tcpInfo],
- p => Bandwidth.TcpEcho(PeakRateHz("tcp_echo", p), p.GetInt("max")));
+ p => Bandwidth.TcpEcho(PeakRateHz("tcp_ping", p), p.GetInt("max")));
- public static readonly KindSpec TcpBulk = new(
- "tcp_bulk",
+ public static readonly KindSpec TcpTransfer = new(
+ "tcp_transfer",
"TCP bulk",
"Bulk TCP transfer on several streams with periodic byte counters and tcp_info.",
Scope.Line,
@@ -110,8 +110,8 @@ public static class Kinds
],
_ => 0);
- public static readonly KindSpec Connect = new(
- "connect",
+ public static readonly KindSpec TcpConnect = new(
+ "tcp_connect",
"Connect",
"Repeated TCP and TLS connection establishment with per-phase timings and outcome.",
Scope.Line | Scope.Target,
@@ -129,8 +129,8 @@ public static class Kinds
],
p => p.GetInt("per_minute") * 4000 / 60);
- public static readonly KindSpec Hold = new(
- "hold",
+ public static readonly KindSpec TlsHold = new(
+ "tls_hold",
"Hold",
"A long-lived TLS connection with periodic heartbeats; records resets and lifetime.",
Scope.Line,
@@ -144,8 +144,38 @@ public static class Kinds
],
p => 2000 / Math.Max(1, p.GetInt("heartbeat_s")) * 8);
- public static readonly KindSpec IcmpEcho = new(
- "icmp_echo",
+ public static readonly KindSpec HttpRequest = new(
+ "http_request",
+ "HTTP request",
+ "One Agent requests a URL over HTTP/1.1, HTTP/2 or HTTP/3 and records DNS, TCP, time to first byte, total time and status. Target is the URL.",
+ Scope.Target,
+ new BlockSet("http", "fixed", "fixed", "request", []),
+ [
+ P.Enum("version", "h2", ["h1", "h2", "h3"], "HTTP version; the request fails instead of falling back."),
+ P.Enum("method", "GET", ["GET", "HEAD"]),
+ P.Bool("adaptive", true, "Adapt the request rate to the failure rate: low while the target answers, up to rate_max_per_min when it fails, exponential back-off while it is unreachable."),
+ P.Int("per_minute", 6, 1, 120, "/min", "Fixed rate, used when adaptive is false."),
+ P.Int("rate_min_per_min", 1, 1, 120, "/min"),
+ P.Int("rate_max_per_min", 12, 1, 120, "/min", "Highest rate, and the reference grid for sample weights."),
+ P.Number("tau_pct", 5, 0.1, 50, "%", "Failure rate a quiet target has to be able to prove below, at 95% confidence."),
+ P.Int("timeout_ms", 10000, 500, 60000, "ms"),
+ P.Int("read_kb", 16, 0, 1024, "KiB", "Body bytes to read before the request counts as finished (0: headers only)."),
+ P.Int("ok_status", 0, 0, 599, null, "Status that counts as success; 0 means any status below 400."),
+ P.Bool("tls_verify", true),
+ ],
+ [
+ new StreamSpec(
+ "req",
+ [
+ Col.Seq("attempt"), Col.TsDialer("ts"), Col.U16("w"),
+ Col.U32("dns_us", nullable: true), Col.U32("tcp_us", nullable: true), Col.U32("ttfb_us", nullable: true), Col.U32("total_us", nullable: true),
+ Col.U16("status", nullable: true), Col.U8("proto", nullable: true), Col.U8("outcome"),
+ ]),
+ ],
+ p => p.GetInt("rate_max_per_min") * (2000 + p.GetInt("read_kb") * 1024) * 8 / 60);
+
+ public static readonly KindSpec IcmpPing = new(
+ "icmp_ping",
"ICMP echo",
"拨号方 or Agent side ICMP echo to a target address.",
Scope.Target,
@@ -159,7 +189,7 @@ public static class Kinds
],
p => Bandwidth.UdpEcho(p.GetInt("rate_hz"), p.GetInt("size")));
- public static readonly IReadOnlyList All = [UdpEcho, UdpTrain, TcpEcho, TcpBulk, Connect, Hold, IcmpEcho];
+ public static readonly IReadOnlyList All = [QuicPing, QuicStream, TcpPing, TcpTransfer, TcpConnect, TlsHold, HttpRequest, IcmpPing];
public static KindSpec? Find(string id)
{
diff --git a/backend/shared/MoShiro.Protocol/Rpc/MoShiroRpc.cs b/backend/shared/MoShiro.Protocol/Rpc/MoShiroRpc.cs
index ee50931..129b238 100644
--- a/backend/shared/MoShiro.Protocol/Rpc/MoShiroRpc.cs
+++ b/backend/shared/MoShiro.Protocol/Rpc/MoShiroRpc.cs
@@ -488,7 +488,7 @@ public sealed class ControlHealth
/// 任务层：测量自己使用的链路。TCP 类积木自带套接字和重拨，QUIC 数据通道由拨号方单独维护。
public sealed class TaskHealth
{
- /// 正在运行的 TCP 类线路积木数（tcp_echo、connect、hold、tcp_bulk）。
+ /// 正在运行的 TCP 类线路积木数（tcp_ping、connect、hold、tcp_transfer）。
[JsonPropertyName("tcp_blocks")]
public int TcpBlocks { get; set; }
@@ -496,7 +496,7 @@ public sealed class TaskHealth
public QuicHealth Quic { get; set; } = new();
}
-/// QUIC 数据通道（udp_echo、udp_train 使用）的状态、握手统计和传输层统计。
+/// QUIC 数据通道（quic_ping、quic_stream 使用）的状态、握手统计和传输层统计。
public sealed class QuicHealth
{
[JsonPropertyName("state")]
diff --git a/backend/shared/MoShiro.Protocol/Tasks/TargetHost.cs b/backend/shared/MoShiro.Protocol/Tasks/TargetHost.cs
new file mode 100644
index 0000000..cfb5eaa
--- /dev/null
+++ b/backend/shared/MoShiro.Protocol/Tasks/TargetHost.cs
@@ -0,0 +1,104 @@
+using System.Globalization;
+using System.Net;
+
+namespace MoShiro.Protocol;
+
+///
+/// 单个 Agent 任务的目标：http 是 http(s)://主机[:端口][/路径]，其余 Kind 是 主机[:端口]（IP 或域名，IPv6 带端口时用方括号）。
+/// Backend 只检查格式，不做 DNS 解析；解析在 Agent 上进行，每个解析出的地址都要在 allowed_target_cidrs 里。
+///
+public static class TargetHost
+{
+ public const string HttpKind = "http_request";
+
+ /// 解析目标； 为 0 表示没有写端口， 只在 http(s) URL 时有值。
+ public static bool TryParse(string? target, out string host, out int port, out Uri? url)
+ {
+ host = "";
+ port = 0;
+ url = null;
+ if (string.IsNullOrWhiteSpace(target) || target.Length > 2048 || target.Any(char.IsWhiteSpace))
+ {
+ return false;
+ }
+
+ if (target.Contains("://", StringComparison.Ordinal))
+ {
+ if (!Uri.TryCreate(target, UriKind.Absolute, out Uri? uri) || uri.Scheme is not ("http" or "https") || uri.UserInfo.Length > 0 || !ValidHost(uri.IdnHost))
+ {
+ return false;
+ }
+
+ url = uri;
+ host = uri.IdnHost.Trim('[', ']');
+ port = uri.IsDefaultPort ? 0 : uri.Port;
+ return true;
+ }
+
+ // IPAddress.TryParse 会把「[v6]:端口」的端口丢掉，带方括号的先按端点解析
+ if (!target.StartsWith('[') && IPAddress.TryParse(target, out IPAddress? ip))
+ {
+ host = ip.ToString();
+ return true;
+ }
+
+ if (IPEndPoint.TryParse(target, out IPEndPoint? endpoint))
+ {
+ if (endpoint.Port is 0)
+ {
+ return false;
+ }
+
+ host = endpoint.Address.ToString();
+ port = endpoint.Port;
+ return true;
+ }
+
+ int colon = target.LastIndexOf(':');
+ string name = colon < 0 ? target : target[..colon];
+ if (colon >= 0 && (!int.TryParse(target.AsSpan(colon + 1), NumberStyles.None, CultureInfo.InvariantCulture, out port) || port is < 1 or > 65535))
+ {
+ return false;
+ }
+
+ if (!ValidHost(name))
+ {
+ return false;
+ }
+
+ host = name.ToLowerInvariant();
+ return true;
+ }
+
+ /// 按 Kind 检查并规范化：http 必须是 http(s) URL，其余 Kind 必须是主机[:端口]；不合格返回 null。
+ public static string? Normalize(string kindId, string? target)
+ {
+ if (!TryParse(target, out string host, out int port, out Uri? url) || (url is not null) != (kindId == HttpKind))
+ {
+ return null;
+ }
+
+ if (url is not null)
+ {
+ return url.AbsoluteUri;
+ }
+
+ string shown = IPAddress.TryParse(host, out IPAddress? ip) && ip.AddressFamily == System.Net.Sockets.AddressFamily.InterNetworkV6 && port != 0 ? "[" + host + "]" : host;
+ return port == 0 ? host : shown + ":" + port.ToString(CultureInfo.InvariantCulture);
+ }
+
+ private static bool ValidHost(string name)
+ {
+ if (IPAddress.TryParse(name.Trim('[', ']'), out _))
+ {
+ return true;
+ }
+
+ if (name.Length is 0 or > 253 || name[^1] == '-' || name[0] is '-' or '.')
+ {
+ return false;
+ }
+
+ return name.TrimEnd('.').Split('.').All(label => label.Length is > 0 and <= 63 && label.All(c => char.IsAsciiLetterOrDigit(c) || c == '-'));
+ }
+}
diff --git a/backend/shared/MoShiro.Protocol/Wire/MeasurementDatagram.cs b/backend/shared/MoShiro.Protocol/Wire/MeasurementDatagram.cs
index 11a44e5..b662387 100644
--- a/backend/shared/MoShiro.Protocol/Wire/MeasurementDatagram.cs
+++ b/backend/shared/MoShiro.Protocol/Wire/MeasurementDatagram.cs
@@ -8,7 +8,7 @@ public enum MeasurementDatagramType : byte
EchoRequest = 1,
EchoReply = 2,
- /// 单向数据（udp_train），以及 拨号方 为了保持 NAT 映射和让 Agent 学到源地址而发的探活包。
+ /// 单向数据（quic_stream），以及 拨号方 为了保持 NAT 映射和让 Agent 学到源地址而发的探活包。
Data = 3,
}
diff --git a/backend/shared/tests/MoShiro.Measurement.Tests/MeasurementTests.cs b/backend/shared/tests/MoShiro.Measurement.Tests/MeasurementTests.cs
index 64b3e46..fe5d99a 100644
--- a/backend/shared/tests/MoShiro.Measurement.Tests/MeasurementTests.cs
+++ b/backend/shared/tests/MoShiro.Measurement.Tests/MeasurementTests.cs
@@ -1,3 +1,4 @@
+using System.Net;
using System.Text.Json;
using Fleet.Core;
using MoShiro.Protocol;
@@ -24,11 +25,11 @@ public sealed class MeasurementTests
BudgetGuard guard = Guard();
double mbps = 0;
long pps = 0;
- Assert.True(guard.Check(Line("udp_echo"), ref mbps, ref pps).Accepted);
+ Assert.True(guard.Check(Line("quic_ping"), ref mbps, ref pps).Accepted);
Assert.Equal(RejectReasons.Unsupported, guard.Check(Line("no_such_kind"), ref mbps, ref pps).Reason);
- Assert.Equal(RejectReasons.InvalidParams, guard.Check(Line("udp_echo", ("rate_hz", "100000")), ref mbps, ref pps).Reason);
- Assert.Equal(RejectReasons.InvalidParams, guard.Check(Line("udp_echo", ("nope", "1")), ref mbps, ref pps).Reason);
- Assert.Equal(RejectReasons.PolicyDenied, guard.Check(Line("udp_echo", ("port", "9999")), ref mbps, ref pps).Reason);
+ Assert.Equal(RejectReasons.InvalidParams, guard.Check(Line("quic_ping", ("rate_hz", "100000")), ref mbps, ref pps).Reason);
+ Assert.Equal(RejectReasons.InvalidParams, guard.Check(Line("quic_ping", ("nope", "1")), ref mbps, ref pps).Reason);
+ Assert.Equal(RejectReasons.PolicyDenied, guard.Check(Line("quic_ping", ("port", "9999")), ref mbps, ref pps).Reason);
}
[Fact]
@@ -37,11 +38,11 @@ public sealed class MeasurementTests
BudgetGuard guard = Guard(new NodeLimits { MaxTotalMbps = 15, AllowedQuicPorts = [7443] });
double mbps = 0;
long pps = 0;
- Assert.True(guard.CheckPhase("udp_train", new Dictionary { ["mbps"] = Json("10") }).Accepted);
- Assert.Equal(RejectReasons.BudgetExceeded, guard.CheckPhase("udp_train", new Dictionary { ["mbps"] = Json("20") }).Reason);
+ Assert.True(guard.CheckPhase("quic_stream", new Dictionary { ["mbps"] = Json("10") }).Accepted);
+ Assert.Equal(RejectReasons.BudgetExceeded, guard.CheckPhase("quic_stream", new Dictionary { ["mbps"] = Json("20") }).Reason);
- Assert.True(guard.Check(Line("udp_train", ("mbps", "10")), ref mbps, ref pps).Accepted);
- Assert.Equal(RejectReasons.BudgetExceeded, guard.Check(Line("udp_train", ("mbps", "10")), ref mbps, ref pps).Reason);
+ Assert.True(guard.Check(Line("quic_stream", ("mbps", "10")), ref mbps, ref pps).Accepted);
+ Assert.Equal(RejectReasons.BudgetExceeded, guard.Check(Line("quic_stream", ("mbps", "10")), ref mbps, ref pps).Reason);
}
[Fact]
@@ -54,8 +55,15 @@ public sealed class MeasurementTests
Assert.True(some.TargetAllowed("192.0.2.17:443"));
Assert.True(some.TargetAllowed("2001:db8::1"));
Assert.False(some.TargetAllowed("192.0.3.1"));
- Assert.False(some.TargetAllowed("example.com"));
Assert.False(some.TargetAllowed(null));
+
+ // 域名要解析之后才知道地址：配置过白名单就先接受，每个解析出的地址再由 AddressAllowed 检查
+ Assert.True(some.TargetAllowed("example.com:443"));
+ Assert.True(some.TargetAllowed("https://example.com/health"));
+ Assert.False(Guard().TargetAllowed("example.com"));
+ Assert.True(some.AddressAllowed(IPAddress.Parse("192.0.2.9")));
+ Assert.False(some.AddressAllowed(IPAddress.Parse("10.0.0.1")));
+ Assert.False(some.AddressAllowed(IPAddress.Parse("::ffff:10.0.0.1")));
}
// ---- RowStore ----
@@ -63,7 +71,7 @@ public sealed class MeasurementTests
[Fact]
public void RowStore_RecordsOnlyItsOwnSidesColumns_AndCutsExistingRowsInKeyOrder()
{
- StreamSpec msg = Kinds.UdpEcho.FindStream("msg")!;
+ StreamSpec msg = Kinds.QuicPing.FindStream("msg")!;
var dialer = new RowStore(msg, Side.Dialer, 64);
Assert.Equal(["size", "w", "t1", "t4"], msg.ColumnsOf(Side.Dialer).Select(c => c.Name));
Assert.Equal(["t2", "t3"], msg.ColumnsOf(Side.Listener).Select(c => c.Name));
@@ -88,7 +96,7 @@ public sealed class MeasurementTests
[Fact]
public void RowStore_RingOverwritesOldRows_AndNeverReturnsAStaleOne()
{
- var store = new RowStore(Kinds.UdpEcho.FindStream("msg")!, Side.Listener, 16);
+ var store = new RowStore(Kinds.QuicPing.FindStream("msg")!, Side.Listener, 16);
store.SetRow(3, [30, 31]);
Assert.True(store.TryGet(3, 0, out long t2));
Assert.Equal(30, t2);
@@ -117,19 +125,19 @@ public sealed class MeasurementTests
msg.Add(seq, 64, 1, t1, t4);
}
- return new RawChunk(new RawChunkHeader { Kind = "udp_echo", Clock = new ChunkClock { DialerOffsetUs = 1000 } }, [msg.Build()]);
+ return new RawChunk(new RawChunkHeader { Kind = "quic_ping", Clock = new ChunkClock { DialerOffsetUs = 1000 } }, [msg.Build()]);
}
[Fact]
public void FragmentMerger_DialerDriven_AddsTheAgentColumnsByKey_AndLeavesMissingRowsEmpty()
{
- var agent = new RowStore(Kinds.UdpEcho.FindStream("msg")!, Side.Listener, 64);
+ var agent = new RowStore(Kinds.QuicPing.FindStream("msg")!, Side.Listener, 64);
agent.SetRow(0, [2000, 2050]);
agent.SetRow(2, [4000, 4040]);
var source = new Source { Stores = { ["msg"] = agent } };
RawChunk fragment = EchoFragment((0, 1000, 3000), (1, 2000, null), (2, 3000, 5000));
- Result> merged = FragmentMerger.Merge(Kinds.UdpEcho, fragment, 1, source);
+ Result> merged = FragmentMerger.Merge(Kinds.QuicPing, fragment, 1, source);
RawStream msg = Assert.Single(merged.Value);
Assert.Equal(["seq", "size", "w", "t1", "t2", "t3", "t4"], msg.Columns.Select(c => c.Name));
Assert.Equal([true, false, true], Enumerable.Range(0, 3).Select(i => msg.Find("t2")!.IsValid(i)));
@@ -141,7 +149,7 @@ public sealed class MeasurementTests
public void FragmentMerger_DialerDriven_WithoutAnyAgentRecords_KeepsTheDialerRows()
{
RawChunk fragment = EchoFragment((0, 1000, 3000), (1, 2000, null));
- RawStream msg = Assert.Single(FragmentMerger.Merge(Kinds.UdpEcho, fragment, 1, new Source()).Value);
+ RawStream msg = Assert.Single(FragmentMerger.Merge(Kinds.QuicPing, fragment, 1, new Source()).Value);
Assert.Equal(2, msg.Count);
Assert.False(msg.Find("t2")!.IsValid(0));
Assert.Equal(3000, msg.Find("t4")!.Values[0]);
@@ -152,14 +160,14 @@ public sealed class MeasurementTests
{
var bad = new StreamBuilder(new StreamSpec("msg", [Col.Seq("seq")]));
bad.Add(0);
- var fragment = new RawChunk(new RawChunkHeader { Kind = "udp_echo" }, [bad.Build()]);
- Assert.False(FragmentMerger.Merge(Kinds.UdpEcho, fragment, 1, new Source()).IsSuccess);
+ var fragment = new RawChunk(new RawChunkHeader { Kind = "quic_ping" }, [bad.Build()]);
+ Assert.False(FragmentMerger.Merge(Kinds.QuicPing, fragment, 1, new Source()).IsSuccess);
}
[Fact]
public void FragmentMerger_AgentDriven_TakesRowsFromTheAgent_AndMarksUnreceivedPacketsAsLost()
{
- var agent = new RowStore(Kinds.UdpTrain.FindStream("pkt")!, Side.Listener, 64);
+ var agent = new RowStore(Kinds.QuicStream.FindStream("pkt")!, Side.Listener, 64);
for (uint seq = 0; seq < 5; seq++)
{
agent.SetRow(seq, [1000, 100 + seq]);
@@ -167,12 +175,12 @@ public sealed class MeasurementTests
var pkt = new StreamBuilder(new StreamSpec("pkt", [Col.Seq("seq"), Col.TsDialer("t_recv")]));
pkt.Add(1, 5000).Add(1, 9999).Add(3, 6000); // sequence 1 arrives twice (the first one counts); 0, 2 and 4 are lost
- var header = new RawChunkHeader { Kind = "udp_train" };
+ var header = new RawChunkHeader { Kind = "quic_stream" };
header.Params[FragmentKeys.SeqFrom] = Json("0");
header.Params[FragmentKeys.SeqTo] = Json("5");
var source = new Source { Stores = { ["pkt"] = agent } };
- RawStream merged = Assert.Single(FragmentMerger.Merge(Kinds.UdpTrain, new RawChunk(header, [pkt.Build()]), 1, source).Value);
+ RawStream merged = Assert.Single(FragmentMerger.Merge(Kinds.QuicStream, new RawChunk(header, [pkt.Build()]), 1, source).Value);
Assert.Equal(5, merged.Count);
Assert.Equal(["seq", "size", "t_send", "t_recv"], merged.Columns.Select(c => c.Name));
RawColumn recv = merged.Find("t_recv")!;
@@ -181,7 +189,7 @@ public sealed class MeasurementTests
Assert.Equal(103, merged.Find("t_send")!.Values[3]);
// A one-way fragment that does not declare its sequence range is invalid
- Assert.False(FragmentMerger.Merge(Kinds.UdpTrain, new RawChunk(new RawChunkHeader { Kind = "udp_train" }, [pkt.Build()]), 1, source).IsSuccess);
+ Assert.False(FragmentMerger.Merge(Kinds.QuicStream, new RawChunk(new RawChunkHeader { Kind = "quic_stream" }, [pkt.Build()]), 1, source).IsSuccess);
}
[Fact]
@@ -189,9 +197,9 @@ public sealed class MeasurementTests
{
var conn = new StreamBuilder(new StreamSpec("conn", [Col.U8("outcome"), Col.Seq("attempt"), Col.TsDialer("ts")]));
conn.Add(0, 0, 100);
- var fragment = new RawChunk(new RawChunkHeader { Kind = "connect" }, [conn.Build()]);
+ var fragment = new RawChunk(new RawChunkHeader { Kind = "tcp_connect" }, [conn.Build()]);
- RawStream merged = Assert.Single(FragmentMerger.Merge(Kinds.Connect, fragment, 1, new Source()).Value);
+ RawStream merged = Assert.Single(FragmentMerger.Merge(Kinds.TcpConnect, fragment, 1, new Source()).Value);
Assert.Equal(["attempt", "ts", "tcp_us", "tls_us", "outcome"], merged.Columns.Select(c => c.Name));
Assert.False(merged.Find("tcp_us")!.IsValid(0));
}
@@ -212,7 +220,7 @@ public sealed class MeasurementTests
[Fact]
public void FragmentMerger_UnionStreams_ConcatenateBothSidesRows_AndEachAgentRowIsDeliveredOnce()
{
- StreamSpec info = Kinds.TcpEcho.FindStream("tcp_info")!;
+ StreamSpec info = Kinds.TcpPing.FindStream("tcp_info")!;
Assert.Equal(info.Columns.Count, info.ColumnsOf(Side.Listener).Count); // each side owns every column
var agent = new RowStore(info, Side.Listener, 64);
var recorder = new Source { Stores = { ["tcp_info"] = agent } };
@@ -222,7 +230,7 @@ public sealed class MeasurementTests
var msg = EchoFragment((0, 1000, 3000));
var fragment = new RawChunk(msg.Header, [msg.Streams[0], InfoRows(info, (100, 0, 1))]);
- IReadOnlyList first = FragmentMerger.Merge(Kinds.TcpEcho, fragment, 1, recorder).Value;
+ IReadOnlyList first = FragmentMerger.Merge(Kinds.TcpPing, fragment, 1, recorder).Value;
RawStream merged = first.Single(x => x.Name == "tcp_info");
Assert.Equal(3, merged.Count);
Assert.Equal([0L, 1L, 1L], merged.Find("side")!.Values);
@@ -231,20 +239,20 @@ public sealed class MeasurementTests
Assert.Equal(1, merged.Find("conn")!.Values[1]);
// The agent's rows were consumed: the next chunk carries only the dialer's own rows
- IReadOnlyList second = FragmentMerger.Merge(Kinds.TcpEcho, fragment, 1, recorder).Value;
+ IReadOnlyList second = FragmentMerger.Merge(Kinds.TcpPing, fragment, 1, recorder).Value;
Assert.Equal(1, second.Single(x => x.Name == "tcp_info").Count);
}
[Fact]
public void FragmentMerger_UnionStreams_AgentRowsAreDeliveredEvenWhenTheDialerHasNone()
{
- StreamSpec info = Kinds.TcpEcho.FindStream("tcp_info")!;
+ StreamSpec info = Kinds.TcpPing.FindStream("tcp_info")!;
var agent = new RowStore(info, Side.Listener, 64);
agent.SetRow(agent.Append(), [500, 1, 40_000, 5_000, 10, 7, 100, 1_000, 1]);
// A dialer on a platform without kernel sampling sends the echo rows only
RawChunk fragment = EchoFragment((0, 1000, 3000));
- IReadOnlyList merged = FragmentMerger.Merge(Kinds.TcpEcho, fragment, 1, new Source { Stores = { ["tcp_info"] = agent } }).Value;
+ IReadOnlyList merged = FragmentMerger.Merge(Kinds.TcpPing, fragment, 1, new Source { Stores = { ["tcp_info"] = agent } }).Value;
RawStream rows = merged.Single(x => x.Name == "tcp_info");
Assert.Equal(1, rows.Count);
Assert.Equal(7, rows.Find("retrans_total")!.Values[0]);
@@ -253,11 +261,11 @@ public sealed class MeasurementTests
[Fact]
public void FragmentMerger_LegacyTcpSamples_KeepConnectionIdentityUnknown()
{
- StreamSpec info = Kinds.TcpEcho.FindStream("tcp_info")!;
+ StreamSpec info = Kinds.TcpPing.FindStream("tcp_info")!;
RawStream sample = InfoRows(info, (100, 0, 1));
var legacy = new RawStream(sample.Name, sample.Count, sample.Columns.Where(c => c.Name != "conn").ToArray());
- var fragment = new RawChunk(new RawChunkHeader { Kind = "tcp_echo" }, [legacy]);
- RawStream merged = Assert.Single(FragmentMerger.Merge(Kinds.TcpEcho, fragment, 1, new Source()).Value);
+ var fragment = new RawChunk(new RawChunkHeader { Kind = "tcp_ping" }, [legacy]);
+ RawStream merged = Assert.Single(FragmentMerger.Merge(Kinds.TcpPing, fragment, 1, new Source()).Value);
Assert.False(merged.Find("conn")!.IsValid(0));
Assert.Equal(1, merged.Find("retrans_total")!.Values[0]);
}
@@ -282,7 +290,7 @@ public sealed class MeasurementTests
[Fact]
public void EchoRows_Tcp_WaitsForTheReplyInsteadOfCallingASeveralSecondStallALoss()
{
- var rows = new EchoRows(Kinds.TcpEcho, 64, EchoRows.TcpGraceUs, settleOnReply: true);
+ var rows = new EchoRows(Kinds.TcpPing, 64, EchoRows.TcpGraceUs, settleOnReply: true);
uint a = rows.Add(64, 0);
uint b = rows.Add(64, 100_000);
uint c = rows.Add(64, 200_000);
@@ -312,7 +320,7 @@ public sealed class MeasurementTests
[Fact]
public void EchoRows_Udp_GivesUpOnAMissingReplyAfterTwoSeconds()
{
- var rows = new EchoRows(Kinds.UdpEcho, 64, EchoRows.DatagramGraceUs, settleOnReply: false);
+ var rows = new EchoRows(Kinds.QuicPing, 64, EchoRows.DatagramGraceUs, settleOnReply: false);
rows.Add(64, 0);
rows.Add(64, 1_000_000);
RawStream cut = rows.Take(2_500_000, all: false)!;
@@ -324,7 +332,7 @@ public sealed class MeasurementTests
[Fact]
public void EchoRows_DiscardedLocalSendIsNotReportedAsNetworkLoss()
{
- var rows = new EchoRows(Kinds.UdpEcho, 64, EchoRows.DatagramGraceUs, settleOnReply: false);
+ var rows = new EchoRows(Kinds.QuicPing, 64, EchoRows.DatagramGraceUs, settleOnReply: false);
uint unsent = rows.Add(64, 0);
Assert.True(rows.Discard(unsent));
@@ -369,7 +377,7 @@ public sealed class MeasurementTests
public void BlockSpec_ConfigSignatureIgnoresTheStreamId_ButSignatureDoesNot()
{
BlockSpec Create(string json, uint stream = 1, int port = 7443) => BlockSpec.Create(
- "udp_echo",
+ "quic_ping",
JsonDocument.Parse(json).RootElement.EnumerateObject().ToDictionary(p => p.Name, p => p.Value.Clone()),
stream,
port,
@@ -381,7 +389,7 @@ public sealed class MeasurementTests
Assert.NotEqual(baseline.Signature, Create("""{"rate_hz":10,"size":64}""", stream: 2).Signature);
Assert.Equal(baseline.ConfigSignature, Create("""{"rate_hz":10,"size":64}""", stream: 2).ConfigSignature);
Assert.NotEqual(baseline.ConfigSignature, Create("""{"rate_hz":10,"size":64}""", port: 9000).ConfigSignature);
- Assert.Null(BlockSpec.Create("udp_echo", new Dictionary { ["rate_hz"] = Json("100000") }, 1, 7443, new FragmentScope(null, null, null)));
+ Assert.Null(BlockSpec.Create("quic_ping", new Dictionary { ["rate_hz"] = Json("100000") }, 1, 7443, new FragmentScope(null, null, null)));
}
// ---- 起止时间 ----
@@ -389,13 +397,13 @@ public sealed class MeasurementTests
[Fact]
public void ChunkFinalizer_TakesStartAndDurationFromTheDataTimestamps_NotFromTheDeliveryTime()
{
- var agent = new RowStore(Kinds.UdpEcho.FindStream("msg")!, Side.Listener, 64);
+ var agent = new RowStore(Kinds.QuicPing.FindStream("msg")!, Side.Listener, 64);
agent.SetRow(0, [2000, 2050]);
RawChunk fragment = EchoFragment((0, 1000, 3000));
- IReadOnlyList streams = FragmentMerger.Merge(Kinds.UdpEcho, fragment, 1, new Source { Stores = { ["msg"] = agent } }).Value;
+ IReadOnlyList streams = FragmentMerger.Merge(Kinds.QuicPing, fragment, 1, new Source { Stores = { ["msg"] = agent } }).Value;
var anchor = new ClockAnchor(ListenerMonoUs: 1_000_000, UtcMs: 1_700_000_000_000);
- RawChunk chunk = ChunkFinalizer.Finalize(Kinds.UdpEcho, fragment, streams, new NodeId("agt_x"), anchor, new Dictionary { ["port"] = Json("7443") }, 7443, new ChunkId(Guid.NewGuid()));
+ RawChunk chunk = ChunkFinalizer.Finalize(Kinds.QuicPing, fragment, streams, new NodeId("agt_x"), anchor, new Dictionary { ["port"] = Json("7443") }, 7443, new ChunkId(Guid.NewGuid()));
// t1 = 1000 us (拨号方 clock) + offset 1000 = 2000 us on the Agent clock, which is 998 ms before the anchor
Assert.Equal(1_700_000_000_000 - 998, chunk.Header.StartUtcMs);
diff --git a/backend/shared/tests/MoShiro.Measurement.Tests/QuicSettingsTests.cs b/backend/shared/tests/MoShiro.Measurement.Tests/QuicSettingsTests.cs
new file mode 100644
index 0000000..4e6159b
--- /dev/null
+++ b/backend/shared/tests/MoShiro.Measurement.Tests/QuicSettingsTests.cs
@@ -0,0 +1,57 @@
+using System.Net;
+using System.Security.Cryptography;
+using System.Security.Cryptography.X509Certificates;
+using Fleet.Core;
+
+namespace MoShiro.Measurement.Tests;
+
+public sealed class QuicSettingsTests
+{
+ [Theory]
+ [InlineData(1100, null, false)]
+ [InlineData(null, 70_000, false)]
+ [InlineData(1400, 1300, false)]
+ [InlineData(1280, 1500, true)]
+ [InlineData(null, null, true)]
+ public void Validate_RejectsMtuOutsideTheSupportedRangeOrOutOfOrder(int? min, int? max, bool valid) =>
+ Assert.Equal(valid, new QuicSettings { MinMtu = min, MaxMtu = max }.Validate() is null);
+
+ /// 设置真的写进了 MsQuic：最大 MTU 压低后数据报上限随之变小，最小 MTU 抬高后一连上就是大包（不用等路径探测）。
+ [Theory]
+ [InlineData(1200, 1280, 1200, 1280)]
+ [InlineData(1500, 1500, 1400, 1500)]
+ public async Task Settings_ChangeTheDatagramLimitOfTheConnection(int min, int max, int atLeast, int atMost)
+ {
+ var settings = new QuicSettings { MinMtu = min, MaxMtu = max };
+ var log = new NullLog();
+ using ECDsa key = ECDsa.Create(ECCurve.NamedCurves.nistP256);
+ var request = new CertificateRequest("CN=localhost", key, HashAlgorithmName.SHA256);
+ using X509Certificate2 certificate = request.CreateSelfSigned(DateTimeOffset.UnixEpoch, DateTimeOffset.UnixEpoch.AddYears(100));
+ await using var listener = new MsQuicListener(IPAddress.Loopback, [0], certificate, (_, _) => { }, log, null, settings);
+ listener.Start();
+ using var timeout = new CancellationTokenSource(TimeSpan.FromSeconds(15));
+ MsQuicConnection client = await MsQuicConnection.ConnectAsync(IPAddress.Loopback, listener.BoundPorts[0], 0, key.ExportSubjectPublicKeyInfo(), (_, _) => { }, log, timeout.Token, settings);
+ try
+ {
+ while (!client.IsReady)
+ {
+ await Task.Delay(10, timeout.Token);
+ }
+
+ // 路径 MTU 探测还可能在往上走，给它一点时间；上限始终不能超过设置的最大 MTU
+ await Task.Delay(1500, timeout.Token);
+ Assert.InRange(client.MaxDatagramBytes, atLeast - 100, atMost);
+ }
+ finally
+ {
+ await client.DisposeAsync().AsTask().WaitAsync(TimeSpan.FromSeconds(3));
+ }
+ }
+
+ private sealed class NullLog : INodeLog
+ {
+ public void Write(NodeLogLevel level, string message, params (string Key, object? Value)[] fields)
+ {
+ }
+ }
+}
diff --git a/backend/shared/tests/MoShiro.Protocol.Tests/CatalogTests.cs b/backend/shared/tests/MoShiro.Protocol.Tests/CatalogTests.cs
index 22e8e51..7f9fa65 100644
--- a/backend/shared/tests/MoShiro.Protocol.Tests/CatalogTests.cs
+++ b/backend/shared/tests/MoShiro.Protocol.Tests/CatalogTests.cs
@@ -35,7 +35,7 @@ public sealed class CatalogTests
public void DefaultLineTemplates_StayWithinTheLineBudget()
{
long total = 0;
- foreach (KindSpec kind in new[] { Kinds.UdpEcho, Kinds.TcpEcho })
+ foreach (KindSpec kind in new[] { Kinds.QuicPing, Kinds.TcpPing })
{
total += kind.BudgetBitsPerSecond(ParamResolver.Resolve(kind.Params, null).Values!);
}
@@ -52,7 +52,7 @@ public sealed class CatalogTests
["size"] = Json("\"big\""),
["bogus"] = Json("1"),
};
- ParamResolution result = ParamResolver.Resolve(Kinds.UdpEcho.Params, input);
+ ParamResolution result = ParamResolver.Resolve(Kinds.QuicPing.Params, input);
Assert.False(result.IsValid);
Assert.Contains("between 1 and 200", result.Errors["rate_hz"]);
@@ -64,13 +64,13 @@ public sealed class CatalogTests
[Fact]
public void Resolve_AcceptsBoundariesAndRejectsJustOutside()
{
- Assert.True(ParamResolver.Resolve(Kinds.UdpEcho.Params, new Dictionary { ["rate_hz"] = Json("1") }).IsValid);
- Assert.True(ParamResolver.Resolve(Kinds.UdpEcho.Params, new Dictionary { ["rate_hz"] = Json("200") }).IsValid);
- Assert.False(ParamResolver.Resolve(Kinds.UdpEcho.Params, new Dictionary { ["rate_hz"] = Json("0") }).IsValid);
- Assert.False(ParamResolver.Resolve(Kinds.UdpEcho.Params, new Dictionary { ["rate_hz"] = Json("201") }).IsValid);
- Assert.False(ParamResolver.Resolve(Kinds.UdpEcho.Params, new Dictionary { ["rate_hz"] = Json("1.5") }).IsValid);
- Assert.False(ParamResolver.Resolve(Kinds.UdpTrain.Params, new Dictionary { ["direction"] = Json("\"sideways\"") }).IsValid);
- Assert.False(ParamResolver.Resolve(Kinds.TcpEcho.Params, new Dictionary { ["tls"] = Json("1") }).IsValid);
+ Assert.True(ParamResolver.Resolve(Kinds.QuicPing.Params, new Dictionary { ["rate_hz"] = Json("1") }).IsValid);
+ Assert.True(ParamResolver.Resolve(Kinds.QuicPing.Params, new Dictionary { ["rate_hz"] = Json("200") }).IsValid);
+ Assert.False(ParamResolver.Resolve(Kinds.QuicPing.Params, new Dictionary { ["rate_hz"] = Json("0") }).IsValid);
+ Assert.False(ParamResolver.Resolve(Kinds.QuicPing.Params, new Dictionary { ["rate_hz"] = Json("201") }).IsValid);
+ Assert.False(ParamResolver.Resolve(Kinds.QuicPing.Params, new Dictionary { ["rate_hz"] = Json("1.5") }).IsValid);
+ Assert.False(ParamResolver.Resolve(Kinds.QuicStream.Params, new Dictionary { ["direction"] = Json("\"sideways\"") }).IsValid);
+ Assert.False(ParamResolver.Resolve(Kinds.TcpPing.Params, new Dictionary { ["tls"] = Json("1") }).IsValid);
}
[Fact]
@@ -98,7 +98,7 @@ public sealed class CatalogTests
var set = new TaskSet
{
Version = 7,
- Tasks = [new TaskSpec { Id = new TaskId("tsk_1"), Kind = "udp_echo", Params = new Dictionary { ["rate_hz"] = Json("10") } }],
+ Tasks = [new TaskSpec { Id = new TaskId("tsk_1"), Kind = "quic_ping", Params = new Dictionary { ["rate_hz"] = Json("10") } }],
Peers = [new PeerSpec { NodeId = new Fleet.Core.NodeId("agt_1"), PublicKey = "AAAA" }],
};
string text = JsonSerializer.Serialize(set, MoShiroJsonContext.Default.TaskSet);
@@ -114,9 +114,9 @@ public sealed class CatalogTests
[Fact]
public void NodeEnrollMeta_ToleratesUnknownFields_AndKeepsDefaults()
{
- NodeEnrollMeta meta = JsonSerializer.Deserialize("{\"kinds\":[\"udp_echo\"],\"future\":1}", MoShiroJsonContext.Default.NodeEnrollMeta)!;
+ NodeEnrollMeta meta = JsonSerializer.Deserialize("{\"kinds\":[\"quic_ping\"],\"future\":1}", MoShiroJsonContext.Default.NodeEnrollMeta)!;
Assert.Equal(0, meta.TcpPort); // 没有端口表示这个节点不监听
- Assert.Equal(["udp_echo"], meta.Kinds);
+ Assert.Equal(["quic_ping"], meta.Kinds);
}
[Fact]
diff --git a/backend/shared/tests/MoShiro.Protocol.Tests/RawChunkCodecTests.cs b/backend/shared/tests/MoShiro.Protocol.Tests/RawChunkCodecTests.cs
index 54015b3..8c73f5b 100644
--- a/backend/shared/tests/MoShiro.Protocol.Tests/RawChunkCodecTests.cs
+++ b/backend/shared/tests/MoShiro.Protocol.Tests/RawChunkCodecTests.cs
@@ -28,7 +28,7 @@ public sealed class RawChunkCodecTests
private static RawChunk SampleChunk()
{
RawChunk chunk = ChunkFactory.Echo(
- s_agent, "udp_echo", 1_700_000_000_000, 6, 100,
+ s_agent, "quic_ping", 1_700_000_000_000, 6, 100,
i => new EchoPacket(LostUp: i == 2, LostDown: i == 4, 20_000 + (i * 100), 21_000));
chunk.Header.ChunkId = new ChunkId(Guid.Parse("01890000-0000-7000-8000-000000000001"));
return chunk;
@@ -41,7 +41,7 @@ public sealed class RawChunkCodecTests
RawChunk decoded = RawChunkCodec.Decode(RawChunkCodec.Encode(original)).Value;
Assert.Equal(original.Header.ChunkId, decoded.Header.ChunkId);
- Assert.Equal("udp_echo", decoded.Header.Kind);
+ Assert.Equal("quic_ping", decoded.Header.Kind);
Assert.Equal(original.Header.Clock.DialerOffsetUs, decoded.Header.Clock.DialerOffsetUs);
RawStream before = original.Find("msg")!;
RawStream after = decoded.Find("msg")!;
@@ -100,7 +100,7 @@ public sealed class RawChunkCodecTests
string columns = Convert.ToHexStringLower(body.AsSpan(4 + (int)headerLength));
string header = System.Text.Encoding.UTF8.GetString(body.AsSpan(4, (int)headerLength));
- Assert.StartsWith("{\"chunk_id\":\"01890000-0000-7000-8000-000000000001\",\"container_version\":1,\"kind\":\"udp_echo\"", header);
+ Assert.StartsWith("{\"chunk_id\":\"01890000-0000-7000-8000-000000000001\",\"container_version\":1,\"kind\":\"quic_ping\"", header);
Assert.Contains("\"streams\":[{\"name\":\"msg\",\"count\":6,\"columns\":[{\"name\":\"seq\",\"type\":\"u32\",\"enc\":\"delta\",\"nullable\":false},", header);
Assert.Equal(GoldenColumns, columns);
}
@@ -181,7 +181,7 @@ public sealed class RawChunkCodecTests
[Fact]
public void Encode_IsSmallerThanTheNaiveRepresentation()
{
- // 10 Hz 的 udp_echo 一个 10 秒块应当远小于每行 6×8 字节
+ // 10 Hz 的 quic_ping 一个 10 秒块应当远小于每行 6×8 字节
RawChunk chunk = ChunkFactory.UdpEchoClean(s_agent, 1_700_000_000_000);
Assert.True(RawChunkCodec.Encode(chunk).Length < 100 * 6 * 8 / 4);
}
diff --git a/backend/shared/tests/MoShiro.Protocol.Tests/TargetHostTests.cs b/backend/shared/tests/MoShiro.Protocol.Tests/TargetHostTests.cs
new file mode 100644
index 0000000..fdf6c93
--- /dev/null
+++ b/backend/shared/tests/MoShiro.Protocol.Tests/TargetHostTests.cs
@@ -0,0 +1,28 @@
+using MoShiro.Protocol;
+
+namespace MoShiro.Protocol.Tests;
+
+public sealed class TargetHostTests
+{
+ [Theory]
+ [InlineData("http_request", "https://Example.com/health?x=1", "https://example.com/health?x=1")]
+ [InlineData("http_request", "http://192.0.2.1:8080", "http://192.0.2.1:8080/")]
+ [InlineData("tcp_connect", "Example.com:443", "example.com:443")]
+ [InlineData("tcp_connect", "192.0.2.1", "192.0.2.1")]
+ [InlineData("tcp_connect", "[2001:db8::1]:443", "[2001:db8::1]:443")]
+ [InlineData("icmp_ping", "2001:db8::1", "2001:db8::1")]
+ public void Normalize_AcceptsTheTargetFormOfTheKind(string kind, string target, string expected) =>
+ Assert.Equal(expected, TargetHost.Normalize(kind, target));
+
+ [Theory]
+ [InlineData("http_request", "example.com")] // http 必须是 URL
+ [InlineData("http_request", "ftp://example.com/")]
+ [InlineData("http_request", "https://user:pw@example.com/")]
+ [InlineData("tcp_connect", "https://example.com/")] // 其余 Kind 不能是 URL
+ [InlineData("tcp_connect", "example.com:0")]
+ [InlineData("tcp_connect", "exa mple.com")]
+ [InlineData("tcp_connect", "-bad.example.com")]
+ [InlineData("tcp_connect", "")]
+ public void Normalize_RejectsTheWrongFormForTheKind(string kind, string target) =>
+ Assert.Null(TargetHost.Normalize(kind, target));
+}