backend: store adaptive rate changes as events and serve an audit of them per line

2026-10-22 10:44:46 +08:00
1 parent d63c3bff74
commit 9ad34acbac
@@ -99,6 +99,7 @@ public static class ApiModule
        RouteGroupBuilder publicGroup = app.MapGroup("/api/public").AddEndpointFilter(PublicGuard.InvokeAsync);
        QueryEndpoints.Map(publicGroup);
        TargetEndpoints.Map(publicGroup);
+        AdaptiveEndpoints.Map(publicGroup);
        LiveEndpoints.Map(publicGroup);
        HostEndpoints.Map(publicGroup);
        NodeEndpoints.Map(publicGroup);
@@ -18,6 +18,7 @@ namespace MoShiro.Backend.Api;
 [JsonSerializable(typeof(IdDto))]
 [JsonSerializable(typeof(OverviewDto))]
 [JsonSerializable(typeof(SeriesDto))]
+[JsonSerializable(typeof(AdaptiveDto))]
 [JsonSerializable(typeof(CompareDto))]
 [JsonSerializable(typeof(DistributionDto))]
 [JsonSerializable(typeof(HeatmapDto))]
@@ -240,3 +240,13 @@ public sealed record TargetDto(
    IReadOnlyList<long> T,
    IReadOnlyDictionary<string, double?[]> Values,
    SummaryDto Summary);
+
+/// <summary>一个时间桶里一种采集方法的自适应探测频率：各状态的秒数、时间加权的平均频率（Hz，不知道时为 null）、状态变化次数、变化时的最大丢包率（%）。</summary>
+public sealed record AdaptiveBucketDto(DateTimeOffset Start, double QuietSeconds, double WatchSeconds, double ActiveSeconds, double DeadSeconds, double? AverageHz, int Changes, double MaxLossPct);
+
+public sealed record AdaptiveKindDto(string Kind, IReadOnlyList<AdaptiveBucketDto> Buckets);
+
+/// <summary>自适应探测频率的一次状态变化：从哪个状态到哪个状态、之后的频率、原因，以及当时的丢包率和包数。</summary>
+public sealed record AdaptiveChangeDto(DateTimeOffset At, string Kind, string From, string To, double Hz, string Reason, double LossPct, int Sent, int Lost);
+
+public sealed record AdaptiveDto(string LineId, DateTimeOffset From, DateTimeOffset To, string Step, IReadOnlyList<AdaptiveKindDto> Kinds, IReadOnlyList<AdaptiveChangeDto> Changes);
@@ -0,0 +1,68 @@
+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;
+
+/// <summary>自适应探测频率的审计：一条线路在一段时间里每种采集方法各状态占多久、平均频率多少，以及每一次状态变化和原因。</summary>
+public static class AdaptiveEndpoints
+{
+    private const string Tag = "public";
+    private const long MaxRangeMs = 30 * Steps.Day;
+    private const int MaxBuckets = 1000;
+
+    public static void Map(RouteGroupBuilder group) =>
+        group.MapGet("/lines/{line_id}/adaptive", AdaptiveAsync)
+            .Api("lines_adaptive", "Adaptive probe-rate states of one line over time (audit)", Tag)
+            .Returns<AdaptiveDto>().ProducesProblems(403, 404, 422, 503);
+
+    private static Task<IResult> AdaptiveAsync(
+        HttpContext http, LineService lines, AdaptiveAuditService audit, TimeProvider time, string line_id, string? range, DateTimeOffset? from, DateTimeOffset? to, string? step) =>
+        QueryHelpers.GuardedAsync(http, async ct =>
+        {
+            Result<LineView> found = await QueryHelpers.FindVisibleLineAsync(http, lines, line_id, ct);
+            if (!found.TryGet(out LineView? view, out FleetError? error))
+            {
+                return Problems.From(http, error);
+            }
+
+            long now = time.GetUtcNow().ToUnixTimeMilliseconds();
+            long toMs = to?.ToUnixTimeMilliseconds() ?? now;
+            long spanMs;
+            if (from is { } start)
+            {
+                spanMs = toMs - start.ToUnixTimeMilliseconds();
+            }
+            else if (!RangeResolver.TryParseSpan(range ?? "48h", out spanMs))
+            {
+                return Problems.From(http, ValidationError.Single("range", "must look like 48h or 7d"));
+            }
+
+            if (spanMs <= 0 || spanMs > MaxRangeMs)
+            {
+                return Problems.From(http, ValidationError.Single("range", "must be between 1 minute and 30 days"));
+            }
+
+            long stepMs = Steps.Hour;
+            if (step is not null && (!RangeResolver.TryParseSpan(step, out stepMs) || stepMs < Steps.Minute))
+            {
+                return Problems.From(http, ValidationError.Single("step", "must look like 15m or 1h"));
+            }
+
+            if (spanMs / stepMs > MaxBuckets)
+            {
+                return Problems.From(http, ValidationError.Single("step", $"too fine for this range (at most {MaxBuckets} buckets)"));
+            }
+
+            long fromMs = toMs - spanMs;
+            AdaptiveResult result = await audit.GetAsync(view.Key, fromMs, toMs, stepMs, ct);
+            var dto = new AdaptiveDto(
+                view.Key.Id, Mappers.Time(result.FromMs), Mappers.Time(result.ToMs), Steps.Format(result.StepMs),
+                [.. result.Kinds.Select(k => new AdaptiveKindDto(k.Kind, [.. k.Buckets.Select(b => new AdaptiveBucketDto(Mappers.Time(b.StartMs), b.QuietSeconds, b.WatchSeconds, b.ActiveSeconds, b.DeadSeconds, b.AverageHz, b.Changes, b.MaxLossPct))]))],
+                [.. result.Changes.Select(c => new AdaptiveChangeDto(Mappers.Time(c.AtMs), c.Kind, c.From, c.To, c.Hz, c.Reason, c.LossPct, c.Sent, c.Lost))]);
+            return TypedResults.Json(dto, ApiJsonContext.Default.AdaptiveDto);
+        });
+}
@@ -108,6 +108,22 @@ public sealed class MoShiroNodeEventSink : INodeEventSink
        }
    }

+    /// <summary>
+    /// 自适应探测频率的状态变化存成事件（<see cref="AdaptiveAudit.EventKind"/>）：线路任务记在监听方名下（和线路的数据同一个归属），
+    /// 单个 Agent 任务记在上报的节点名下。时间以节点上报的为准，明显不对（未来、或太久以前）的用心跳的时间。
+    /// </summary>
+    private async ValueTask RecordRateChangesAsync(NodeId reporter, long heartbeatMs, List<RateChange> changes, CancellationToken ct)
+    {
+        foreach (RateChange change in changes.Take(256))
+        {
+            change.Dialer = reporter.Value;
+            long at = change.AtMs > heartbeatMs - 3_600_000 && change.AtMs <= heartbeatMs + 60_000 ? change.AtMs : heartbeatMs;
+            change.AtMs = at;
+            NodeId owner = change.Listener is { Length: > 0 } listener ? new NodeId(listener) : reporter;
+            await _events.AddAsync(owner, AdaptiveAudit.EventKind, at, System.Text.Json.JsonSerializer.Serialize(change, MoShiroJsonContext.Default.RateChange), ct);
+        }
+    }
+
    private async ValueTask ApplyHeartbeatAsync(NodeId id, NodeEvent nodeEvent, System.Text.Json.JsonElement extra, CancellationToken ct)
    {
        await _nodes.SetVersionAsync(id.Value, nodeEvent.Node.NodeVersion, ct);
@@ -129,6 +145,11 @@ public sealed class MoShiroNodeEventSink : INodeEventSink
            }

            _nodeLogs.Observe(id, beat.LogSummary);
+            if (beat.RateChanges is { Count: > 0 } rateChanges)
+            {
+                await RecordRateChangesAsync(id, nodeEvent.AtMs, rateChanges, ct);
+            }
+
            state.ReplaceDialers(beat.Dialers.Select(d => new ConnectedDialerInfo(d.Id.Value, d.Version, d.Remote, d.SinceUtcMs, d.IpVersion)));
            foreach (ConnectedDialer dialer in beat.Dialers)
            {
@@ -33,6 +33,7 @@ public static class ApplicationServiceCollectionExtensions
        services.AddSingleton<LinkService>();
        services.AddSingleton<LineService>();
        services.AddSingleton<TargetService>();
+        services.AddSingleton<AdaptiveAuditService>();
        services.AddSingleton<TestService>();
        services.AddSingleton<IngestService>();
        services.AddSingleton<WindowComputer>();
@@ -0,0 +1,155 @@
+using System.Text.Json;
+using Fleet.Core;
+using MoShiro.Protocol;
+
+namespace MoShiro.Backend.Application;
+
+/// <summary>一个时间桶里一种采集方法的自适应探测频率：各状态各占多少秒、时间加权的平均频率、状态变化次数、变化时的最大丢包率。</summary>
+public sealed record AdaptiveBucket(long StartMs, double QuietSeconds, double WatchSeconds, double ActiveSeconds, double DeadSeconds, double? AverageHz, int Changes, double MaxLossPct);
+
+public sealed record AdaptiveKind(string Kind, IReadOnlyList<AdaptiveBucket> Buckets);
+
+public sealed record AdaptiveResult(long FromMs, long ToMs, long StepMs, IReadOnlyList<AdaptiveKind> Kinds, IReadOnlyList<RateChange> Changes);
+
+/// <summary>
+/// 自适应探测频率的审计：节点把每一次状态变化（Quiet / Watch / Active / Dead、频率、原因、丢包）随心跳上报，Backend 存成事件（保留期和别的事件一样），
+/// 这里把它们还原成「某个时段里各状态占多久、平均频率多少」，所以几天之后也能回答「晚高峰的时候自适应是什么状态」。
+/// </summary>
+public sealed class AdaptiveAuditService
+{
+    private readonly IEventStore _events;
+
+    public AdaptiveAuditService(IEventStore events)
+    {
+        _events = events;
+    }
+
+    public async ValueTask<AdaptiveResult> GetAsync(LineKey line, long fromMs, long toMs, long stepMs, CancellationToken ct)
+    {
+        // 往前多取一段：区间开头的状态要由这之前最近的一次变化决定
+        IReadOnlyList<EventRecord> records = await _events.ListAsync(line.Listener, AdaptiveAudit.EventKind, fromMs - AdaptiveAudit.LookBackMs, toMs, null, 20_000, ct);
+        List<RateChange> changes = [];
+        foreach (EventRecord record in records.OrderBy(r => r.AtMs).ThenBy(r => r.Id))
+        {
+            RateChange? change = AdaptiveAudit.Parse(record);
+            if (change is not null && change.Dialer == line.Dialer.Value && (change.IpVersion ?? "ipv4") == line.IpVersion && change.Listener == line.Listener.Value)
+            {
+                changes.Add(change);
+            }
+        }
+
+        return AdaptiveAudit.Summarize(changes, fromMs, toMs, stepMs);
+    }
+}
+
+public static class AdaptiveAudit
+{
+    public const string EventKind = "rate_change";
+
+    /// <summary>区间开头之前最多往回看多久，找那时的状态。</summary>
+    public const long LookBackMs = 24 * 3_600_000L;
+
+    public static RateChange? Parse(EventRecord record)
+    {
+        if (record.Detail is null)
+        {
+            return null;
+        }
+
+        try
+        {
+            RateChange? change = JsonSerializer.Deserialize(record.Detail, MoShiroJsonContext.Default.RateChange);
+            if (change is not null)
+            {
+                change.AtMs = record.AtMs;
+            }
+
+            return change;
+        }
+        catch (JsonException)
+        {
+            return null;
+        }
+    }
+
+    /// <summary>
+    /// 把按时间排好的状态变化还原成每个时间桶里各状态的秒数。一次变化之后到下一次变化之前，是变化后的状态和频率；
+    /// 区间里第一次变化之前，状态是这次变化的「从」（频率不知道，不计入平均）。一种采集方法一条时间线（同一条线路上每种方法各自变化）。
+    /// </summary>
+    public static AdaptiveResult Summarize(IReadOnlyList<RateChange> changes, long fromMs, long toMs, long stepMs)
+    {
+        int count = (int)Math.Max(1, Math.Ceiling((toMs - fromMs) / (double)stepMs));
+        var kinds = new List<AdaptiveKind>();
+        foreach (IGrouping<string, RateChange> group in changes.GroupBy(c => c.Kind).OrderBy(g => g.Key, StringComparer.Ordinal))
+        {
+            var quiet = new double[count];
+            var watch = new double[count];
+            var active = new double[count];
+            var dead = new double[count];
+            var hzWeighted = new double[count];
+            var hzSeconds = new double[count];
+            var changeCounts = new int[count];
+            var maxLoss = new double[count];
+            RateChange[] ordered = [.. group.OrderBy(c => c.AtMs)];
+
+            // 每一段：[开始, 结束) 内的状态；第一段在第一次变化之前，状态取它的「从」
+            string state = ordered[0].From;
+            double? hz = null;
+            long cursor = Math.Min(ordered[0].AtMs, fromMs);
+            foreach (RateChange change in ordered)
+            {
+                AddSegment(cursor, change.AtMs, state, hz);
+                int slot = (int)((change.AtMs - fromMs) / stepMs);
+                if (change.AtMs >= fromMs && change.AtMs < toMs && slot >= 0 && slot < count)
+                {
+                    changeCounts[slot]++;
+                    maxLoss[slot] = Math.Max(maxLoss[slot], change.LossPct);
+                }
+
+                cursor = change.AtMs;
+                state = change.To;
+                hz = change.Hz;
+            }
+
+            AddSegment(cursor, toMs, state, hz);
+            var buckets = new List<AdaptiveBucket>(count);
+            for (int i = 0; i < count; i++)
+            {
+                buckets.Add(new AdaptiveBucket(fromMs + (i * stepMs), Math.Round(quiet[i], 1), Math.Round(watch[i], 1), Math.Round(active[i], 1), Math.Round(dead[i], 1),
+                    hzSeconds[i] > 0 ? Math.Round(hzWeighted[i] / hzSeconds[i], 3) : null, changeCounts[i], maxLoss[i]));
+            }
+
+            kinds.Add(new AdaptiveKind(group.Key, buckets));
+
+            void AddSegment(long start, long end, string segmentState, double? segmentHz)
+            {
+                start = Math.Max(start, fromMs);
+                end = Math.Min(end, toMs);
+                for (long t = start; t < end;)
+                {
+                    int slot = (int)((t - fromMs) / stepMs);
+                    long slotEnd = Math.Min(end, fromMs + ((slot + 1) * stepMs));
+                    double seconds = (slotEnd - t) / 1000.0;
+                    switch (segmentState)
+                    {
+                        case "quiet": quiet[slot] += seconds; break;
+                        case "watch": watch[slot] += seconds; break;
+                        case "active": active[slot] += seconds; break;
+                        case "dead": dead[slot] += seconds; break;
+                    }
+
+                    if (segmentHz is { } known)
+                    {
+                        hzWeighted[slot] += known * seconds;
+                        hzSeconds[slot] += seconds;
+                    }
+
+                    t = slotEnd;
+                }
+            }
+        }
+
+        RateChange[] inRange = [.. changes.Where(c => c.AtMs >= fromMs && c.AtMs < toMs).OrderByDescending(c => c.AtMs).Take(200)];
+        return new AdaptiveResult(fromMs, toMs, stepMs, kinds, inRange);
+    }
+}
@@ -0,0 +1,107 @@
+using System.Text.Json.Nodes;
+using Fleet.Core;
+using MoShiro.Backend.Application;
+using MoShiro.Protocol;
+using MoShiro.TestSupport;
+
+namespace MoShiro.Backend.Tests;
+
+public sealed class AdaptiveAuditTests : IClassFixture<BackendFixture>
+{
+    private static readonly long s_t0 = new DateTimeOffset(2026, 10, 22, 10, 0, 0, TimeSpan.Zero).ToUnixTimeMilliseconds();
+    private const long Minute = 60_000;
+    private const long Hour = 3_600_000;
+
+    private readonly BackendFixture _backend;
+
+    public AdaptiveAuditTests(BackendFixture backend)
+    {
+        _backend = backend;
+    }
+
+    private static RateChange Change(long atMs, string kind, string from, string to, double hz, string reason = "quiet", double lossPct = 0) =>
+        new() { AtMs = atMs, Kind = kind, From = from, To = to, Hz = hz, Reason = reason, LossPct = lossPct, Sent = 1000, Lost = (int)(lossPct * 10) };
+
+    /// <summary>状态变化还原成每个时间桶里各状态的秒数：跨桶的一段要切开，区间开头的状态取第一次变化的「从」，不同方法各算各的。</summary>
+    [Fact]
+    public void Summarize_SplitsStateTimeAcrossBuckets_AndWeightsTheAverageRateByTime()
+    {
+        RateChange[] changes =
+        [
+            Change(s_t0 + (30 * Minute), "quic_ping", "watch", "quiet", 2),                      // 10:00–10:30 watch（频率未知），之后 quiet 2 Hz
+            Change(s_t0 + (90 * Minute), "quic_ping", "quiet", "active", 10, "sustained loss", 0.4), // 11:30 起 active 10 Hz
+            Change(s_t0 + (30 * Minute), "tcp_ping", "quiet", "watch", 20, "rtt above baseline"),
+        ];
+
+        AdaptiveResult result = AdaptiveAudit.Summarize(changes, s_t0, s_t0 + (2 * Hour), Hour);
+
+        AdaptiveKind quic = Assert.Single(result.Kinds, k => k.Kind == "quic_ping");
+        Assert.Equal(2, quic.Buckets.Count);
+        AdaptiveBucket first = quic.Buckets[0];
+        Assert.Equal((1800, 1800, 0), ((int)first.WatchSeconds, (int)first.QuietSeconds, (int)first.ActiveSeconds)); // 10:00–10:30 watch，10:30–11:00 quiet
+        Assert.Equal(2, first.AverageHz);                                                                           // 只有 quiet 那半小时知道频率
+        Assert.Equal(1, first.Changes);
+        AdaptiveBucket second = quic.Buckets[1];
+        Assert.Equal((1800, 1800), ((int)second.QuietSeconds, (int)second.ActiveSeconds));                          // 11:00–11:30 quiet，11:30–12:00 active
+        Assert.Equal(6, second.AverageHz);                                                                          // (2 Hz × 30 min + 10 Hz × 30 min) / 60 min
+        Assert.Equal((1, 0.4), (second.Changes, second.MaxLossPct));
+
+        AdaptiveKind tcp = Assert.Single(result.Kinds, k => k.Kind == "tcp_ping");
+        Assert.Equal((1800, 1800), ((int)tcp.Buckets[0].QuietSeconds, (int)tcp.Buckets[0].WatchSeconds));
+        Assert.Equal(3600, (int)tcp.Buckets[1].WatchSeconds);
+
+        Assert.Equal(3, result.Changes.Count);
+        Assert.Equal("quic_ping", result.Changes[0].Kind); // 最新的在前
+    }
+
+    /// <summary>变化在区间开头之前发生：那之后的状态延续到区间里。</summary>
+    [Fact]
+    public void Summarize_CarriesTheStateInFromBeforeTheRange()
+    {
+        RateChange[] changes = [Change(s_t0 - (2 * Hour), "quic_ping", "quiet", "active", 10, "sustained loss", 1)];
+
+        AdaptiveResult result = Summarize(changes);
+
+        AdaptiveBucket bucket = Assert.Single(Assert.Single(result.Kinds).Buckets);
+        Assert.Equal(3600, (int)bucket.ActiveSeconds);
+        Assert.Equal((10, 0), (bucket.AverageHz, bucket.Changes));
+        Assert.Empty(result.Changes); // 区间里没有变化
+
+        AdaptiveResult Summarize(RateChange[] c) => AdaptiveAudit.Summarize(c, s_t0, s_t0 + Hour, Hour);
+    }
+
+    /// <summary>端到端：拨号节点随心跳上报的状态变化存成事件，几天后还能通过接口按时段查到。</summary>
+    [Fact]
+    public async Task RateChangesReportedInHeartbeats_AreStored_AndServedByTheAdaptiveEndpoint()
+    {
+        await using AgentSimulator first = await _backend.ConnectAgentAsync();
+        await using AgentSimulator second = await _backend.ConnectAgentAsync();
+        (AgentSimulator dialer, AgentSimulator listener) = string.CompareOrdinal(first.NodeId.Value, second.NodeId.Value) < 0 ? (first, second) : (second, first); // 小 id 拨大 id
+        long now = TimeProvider.System.GetUtcNow().ToUnixTimeMilliseconds();
+
+        RateChange Of(long atMs, string from, string to, double hz, string reason, double loss)
+        {
+            RateChange change = Change(atMs, "quic_ping", from, to, hz, reason, loss);
+            change.Listener = listener.NodeId.Value;
+            change.IpVersion = "ipv4";
+            return change;
+        }
+
+        await dialer.SendRateChangesAsync(Of(now - (20 * Minute), "watch", "quiet", 2, "quiet", 0), Of(now - (10 * Minute), "quiet", "active", 10, "sustained loss", 0.5));
+
+        string lineId = listener.NodeId.Value + "~" + dialer.NodeId.Value;
+        JsonNode? audit = null;
+        await BackendFixture.EventuallyAsync(async () =>
+        {
+            audit = await _backend.GetJsonAsync(_backend.Admin, $"/api/public/lines/{lineId}/adaptive?range=1h&step=10m");
+            return audit["changes"]!.AsArray().Count == 2;
+        });
+
+        JsonNode quic = Assert.Single(audit!["kinds"]!.AsArray())!;
+        Assert.Equal("quic_ping", quic["kind"]!.GetValue<string>());
+        double active = quic["buckets"]!.AsArray().Sum(b => b!["active_seconds"]!.GetValue<double>());
+        Assert.InRange(active, 500, 700); // 最近 10 分钟 active
+        JsonNode newest = audit!["changes"]![0]!;
+        Assert.Equal(("quiet", "active", "sustained loss"), (newest["from"]!.GetValue<string>(), newest["to"]!.GetValue<string>(), newest["reason"]!.GetValue<string>()));
+    }
+}
@@ -223,6 +223,13 @@ public sealed class AgentSimulator : IAsyncDisposable
            new HeartbeatParams { Extra = JsonSerializer.SerializeToElement(new NodeHeartbeatExtra { Update = status }, MoShiroJsonContext.Default.NodeHeartbeatExtra) },
            FleetRpcContext.Default.HeartbeatParams));

+    /// <summary>像拨号节点那样，随心跳上报自适应探测频率的状态变化。</summary>
+    public Task SendRateChangesAsync(params RateChange[] changes) =>
+        SendAsync(RpcMessage.Notification(
+            FleetMethods.Heartbeat,
+            new HeartbeatParams { Extra = JsonSerializer.SerializeToElement(new NodeHeartbeatExtra { RateChanges = [.. changes] }, MoShiroJsonContext.Default.NodeHeartbeatExtra) },
+            FleetRpcContext.Default.HeartbeatParams));
+
    public async Task<RpcMessage> RequestRawAsync(string method, JsonElement? parameters = null) => await RequestAsync(method, parameters);

    public async Task<TaskSet> WaitForTaskSetAsync(Func<TaskSet, bool> predicate, TimeSpan? timeout = null)