diff --git a/backend/agent/src/MoShiro.Agent/AgentApp.cs b/backend/agent/src/MoShiro.Agent/AgentApp.cs
index b80ffdf..c7f5659 100644
--- a/backend/agent/src/MoShiro.Agent/AgentApp.cs
+++ b/backend/agent/src/MoShiro.Agent/AgentApp.cs
@@ -45,6 +45,7 @@ public sealed class AgentApp : IDisposable, IAsyncDisposable
private readonly PeerListener _peers;
private readonly PeerHub _hub;
private readonly ChunkAssembler _chunks;
+ private readonly RateChangeLog _rateChanges = new();
private readonly TaskManager _tasks;
private readonly TestRunner _tests;
private readonly FleetLink _link;
@@ -83,7 +84,7 @@ public sealed class AgentApp : IDisposable, IAsyncDisposable
_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, quic, 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, _rateChanges), time, log);
_knock = new KnockGate(identity.PeerIdentity, RosterKeys, time);
_tcp = new TcpAcceptor(bind, config.Listen.TcpPorts, identity.Certificate, _sessions, _counters.Data, _knock);
@@ -101,7 +102,7 @@ public sealed class AgentApp : IDisposable, IAsyncDisposable
MaxBulkMbps = config.Limits.MaxTotalMbps,
};
_chunks = new ChunkAssembler(_streams, _clock, _counters, _outbox, identity.NodeId, time, log);
- var targets = new TargetRunners(_chunks, identity.NodeId, _clock, time, log);
+ var targets = new TargetRunners(_chunks, identity.NodeId, _clock, time, log, _rateChanges);
_tasks = new TaskManager(CreateGuard, _streams, env, targets, () => _datagrams.BoundPorts, () => _tcp.BoundPorts, _trust, acceptLineTasks: _listening);
_hub = new PeerHub(_sessions, _clock, _chunks, _tasks, _uplink, time, log);
_tests = new TestRunner(_tasks, _sessions, _streams, env, _uplink, config.Limits.MaxConcurrentTests, transport => transport == "quic_datagram" ? _datagrams.BoundPorts : _tcp.BoundPorts, time, log);
@@ -445,6 +446,7 @@ public sealed class AgentApp : IDisposable, IAsyncDisposable
DroppedRateLimited = _counters.DroppedRateLimited,
QueueBytes = _outbox.PendingBytes,
Update = updateStatus,
+ RateChanges = _rateChanges.Take(64),
LogSummary = logSummary is null ? null : new LogSummaryDto
{
LastId = logSummary.LastId,
diff --git a/backend/agent/src/MoShiro.Agent/Dial/DialConnection.cs b/backend/agent/src/MoShiro.Agent/Dial/DialConnection.cs
index 5d6dd6a..6434e30 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, QuicSettings Quic, 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, RateChangeLog RateChanges);
///
/// 与一个 Agent 的点对点连接,分两层:
@@ -190,6 +190,12 @@ public sealed class DialConnection : IDisposable
Fragments = shipper,
MaxPacketsPerSecond = _env.Limits.MaxTotalPps,
MaxBulkMbps = _env.Limits.MaxTotalMbps,
+ OnRateChange = change =>
+ {
+ change.Listener = Endpoint.NodeId;
+ change.IpVersion = Endpoint.IpVersion;
+ _env.RateChanges.Add(change);
+ },
};
var session = new DialSession(blockEnv, clock, _outbox, _env);
var rpc = new RpcSession(transport, s_router, session, _env.Time);
diff --git a/backend/agent/src/MoShiro.Agent/RateChangeLog.cs b/backend/agent/src/MoShiro.Agent/RateChangeLog.cs
new file mode 100644
index 0000000..364d259
--- /dev/null
+++ b/backend/agent/src/MoShiro.Agent/RateChangeLog.cs
@@ -0,0 +1,40 @@
+using System.Collections.Concurrent;
+using MoShiro.Protocol;
+
+namespace MoShiro.Agent;
+
+///
+/// 自适应探测频率的状态变化,等着随下一次心跳上报给 Backend(Backend 存成事件,用于事后审计)。
+/// 和普通日志分开存:日志环形缓冲几分钟到几小时就会被覆盖,而这里的变化要能在几天后回看。有界,满了丢最旧的。
+///
+public sealed class RateChangeLog
+{
+ public const int Capacity = 512;
+
+ private readonly ConcurrentQueue _queue = new();
+
+ public void Add(RateChange change)
+ {
+ _queue.Enqueue(change);
+ while (_queue.Count > Capacity && _queue.TryDequeue(out _))
+ {
+ }
+ }
+
+ /// 取走最多 条(最旧的在前);没有返回 null,心跳里就不带这个字段。
+ public List? Take(int max)
+ {
+ List? taken = null;
+ while ((taken?.Count ?? 0) < max)
+ {
+ if (!_queue.TryDequeue(out RateChange? change))
+ {
+ break;
+ }
+
+ (taken ??= []).Add(change);
+ }
+
+ return taken;
+ }
+}
diff --git a/backend/agent/src/MoShiro.Agent/Tasks/TargetRunners.cs b/backend/agent/src/MoShiro.Agent/Tasks/TargetRunners.cs
index 6b14ded..c946c03 100644
--- a/backend/agent/src/MoShiro.Agent/Tasks/TargetRunners.cs
+++ b/backend/agent/src/MoShiro.Agent/Tasks/TargetRunners.cs
@@ -24,9 +24,11 @@ public sealed class TargetRunners
private readonly MonoClock _clock;
private readonly TimeProvider _time;
private readonly INodeLog _log;
+ private readonly RateChangeLog? _rateChanges;
- public TargetRunners(ChunkAssembler chunks, NodeId agent, MonoClock clock, TimeProvider time, INodeLog log)
+ public TargetRunners(ChunkAssembler chunks, NodeId agent, MonoClock clock, TimeProvider time, INodeLog log, RateChangeLog? rateChanges = null)
{
+ _rateChanges = rateChanges;
_chunks = chunks;
_agent = agent;
_clock = clock;
@@ -92,6 +94,7 @@ public sealed class TargetRunners
PeerAddress = address,
Target = spec.Target,
AllowAddress = allow,
+ OnRateChange = _rateChanges is null ? null : _rateChanges.Add,
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)));
diff --git a/backend/shared/MoShiro.Measurement/Blocks/BlockBase.cs b/backend/shared/MoShiro.Measurement/Blocks/BlockBase.cs
index 478b75b..4c5ebd5 100644
--- a/backend/shared/MoShiro.Measurement/Blocks/BlockBase.cs
+++ b/backend/shared/MoShiro.Measurement/Blocks/BlockBase.cs
@@ -97,6 +97,20 @@ public abstract class BlockBase : IBlock
private void LogTransition(ProbeTransition change)
{
+ Env.OnRateChange?.Invoke(new RateChange
+ {
+ AtMs = Env.Time.GetUtcNow().ToUnixTimeMilliseconds(),
+ Task = Spec.Scope.Task?.Value,
+ Kind = Spec.KindId,
+ From = change.From.ToString().ToLowerInvariant(),
+ To = change.To.ToString().ToLowerInvariant(),
+ Hz = Math.Round(change.Hz, 3),
+ Reason = change.Reason,
+ LossPct = Math.Round(change.LossPct, 4),
+ Sent = change.Sent,
+ Lost = change.Lost,
+ });
+
var fields = new (string, object?)[]
{
("stream", Spec.StreamId), ("kind", Spec.KindId), ("from", change.From.ToString().ToLowerInvariant()), ("to", change.To.ToString().ToLowerInvariant()), ("hz", Math.Round(change.Hz, 1)),
diff --git a/backend/shared/MoShiro.Measurement/Core/Contracts.cs b/backend/shared/MoShiro.Measurement/Core/Contracts.cs
index 44519c5..1c5dd3a 100644
--- a/backend/shared/MoShiro.Measurement/Core/Contracts.cs
+++ b/backend/shared/MoShiro.Measurement/Core/Contracts.cs
@@ -245,6 +245,9 @@ public sealed class BlockEnv
/// 单个 Agent 任务:解析出来的地址能不能连(本地 allowed_target_cidrs);null 表示不限制。
public Func? AllowAddress { get; init; }
+ /// 自适应频率的状态变化(用于审计);null 表示不记录。
+ public Action? OnRateChange { get; init; }
+
/// 拨号方:对端 TLS 证书的 SPKI 指纹;null 表示不校验证书(目标任务,目标是任意地址)。
public byte[]? PeerSpkiSha256 { get; init; }
@@ -277,6 +280,7 @@ public sealed class BlockEnv
PeerAddress = PeerAddress,
Target = Target,
AllowAddress = AllowAddress,
+ OnRateChange = OnRateChange,
PeerSpkiSha256 = PeerSpkiSha256,
KnockKey = KnockKey,
Fragments = Fragments,
diff --git a/backend/shared/MoShiro.Protocol/Rpc/MoShiroRpc.cs b/backend/shared/MoShiro.Protocol/Rpc/MoShiroRpc.cs
index 129b238..8b02fd8 100644
--- a/backend/shared/MoShiro.Protocol/Rpc/MoShiroRpc.cs
+++ b/backend/shared/MoShiro.Protocol/Rpc/MoShiroRpc.cs
@@ -342,6 +342,10 @@ public sealed class NodeHeartbeatExtra
/// 日志只报计数(不带文本):自上次心跳以来各级别的条数、最多几个出现最多的类型,以及最新一条的编号;文本由 Backend 用 moshiro.logs.fetch 按需拉取。
[JsonPropertyName("log_summary")]
public LogSummaryDto? LogSummary { get; set; }
+
+ /// 自上次心跳以来自适应探测频率的状态变化(节点上限 512 条,超出丢最旧的)。
+ [JsonPropertyName("rate_changes")]
+ public List? RateChanges { get; set; }
}
/// 某一类日志(消息文本的稳定哈希)在一个级别上出现了多少次。
@@ -357,6 +361,56 @@ public sealed class LogTypeCountDto
public long Count { get; set; }
}
+///
+/// 自适应探测频率的一次状态变化(Quiet / Watch / Active / Dead):节点随心跳上报,Backend 存成事件(rate_change),
+/// 事后可以回看每条线路、每个时段的探测频率为什么升降(例如晚高峰)。线路任务带 listener(对端)和 ip_version,目标任务带 task。
+///
+public sealed class RateChange
+{
+ [JsonPropertyName("at_ms")]
+ public long AtMs { get; set; }
+
+ /// 这条线路的监听方节点 id(线路任务)。
+ [JsonPropertyName("listener")]
+ public string? Listener { get; set; }
+
+ /// 拨号方节点 id:节点上报时不填,由 Backend 按上报者补上。
+ [JsonPropertyName("dialer")]
+ public string? Dialer { get; set; }
+
+ [JsonPropertyName("ip_version")]
+ public string? IpVersion { get; set; }
+
+ /// 单个 Agent 任务的任务 id。
+ [JsonPropertyName("task")]
+ public string? Task { get; set; }
+
+ [JsonPropertyName("kind")]
+ public string Kind { get; set; } = "";
+
+ [JsonPropertyName("from")]
+ public string From { get; set; } = "";
+
+ [JsonPropertyName("to")]
+ public string To { get; set; } = "";
+
+ [JsonPropertyName("hz")]
+ public double Hz { get; set; }
+
+ [JsonPropertyName("reason")]
+ public string Reason { get; set; } = "";
+
+ /// 变化时刻证明窗口内的丢包率(百分比)和发出、丢失的包数。
+ [JsonPropertyName("loss_pct")]
+ public double LossPct { get; set; }
+
+ [JsonPropertyName("sent")]
+ public int Sent { get; set; }
+
+ [JsonPropertyName("lost")]
+ public int Lost { get; set; }
+}
+
public sealed class LogSummaryDto
{
[JsonPropertyName("last_id")]
@@ -826,6 +880,7 @@ public static class FragmentKeys
[JsonSerializable(typeof(LogsFetchParams))]
[JsonSerializable(typeof(LogsFetchResult))]
[JsonSerializable(typeof(LogSummaryDto))]
+[JsonSerializable(typeof(RateChange))]
[JsonSerializable(typeof(UpdateStatus))]
[JsonSerializable(typeof(NodeHeartbeatExtra))]
[JsonSerializable(typeof(NodeHelloMeta))]