# 任务与采集引擎

## 1. 原则

- 只采集原始数据：逐包、逐消息、逐事件，TCP 另加周期性的内核采样；不计算任何统计，解释全部放在 [analysis.md](analysis.md)。
- 一处声明，处处生效：采集方法（Kind）由积木组合声明而成。Agent 共用同一套积木（`backend/shared/MoShiro.Measurement`）。
- 密度即参数：频率、包大小、采样间隔全部是任务参数，由 API 下发，不需要改代码。

## 2. 任务模型

同一个 Agent 节点 的 IPv4、IPv6 路径并行执行相同的持续线路任务模板，各自记录、上传、分析。IPv6 只在管理端配置了地址且 Agent 上报支持时建立；两端的控制和数据通道固定使用同一地址族。按需测试通过所选线路 id 指定地址族，重负载并发限制仍按 Agent 的本机策略执行。两个地址族分别使用有界发件箱（拨号方 每路径 8 MB，Agent 的共同上传队列 64 MB）。

```text
Task = { id, scope, kind, target, params, schedule, enabled }

scope     line   : 拨号方 ↔ Agent 线路任务      target 为空（agent_id 为空时表示对所有 Agent 生效的模板）
          target : Agent → 目标 IP 的单边任务   target = ip[:port]（受 Agent 本地 allowed_target_cidrs 限制）
kind      Kind 目录中的 id，例如 udp_echo / tcp_echo / udp_train …
params    按 Kind 的 ParamSpec 校验（默认值、上下限）；未指定的参数取默认值
schedule  continuous   持续运行，按时间切块
          every(N)     每 N 分钟运行一次（N ≥ 60，适用于重负载 Kind）
          once         按需测试，由 API 触发，也就是一次 Test
```

- 线路任务模板：`agent_id` 为空的 `line` 任务会自动应用到所有 `active` 的 Agent。新 Agent 注册后自动获得这些任务；也可以针对单个 Agent 覆盖参数或停用。
- 测试方案（Plan）：多个阶段的 `once` 任务组合，例如 `hy2_ladder` 是 `udp_train` 在 10、25、50、100 Mbps 下各运行 10 秒。Plan 定义在 `backend/src/MoShiro.Backend.Application/Tests/Plans.cs`。
- 期望任务集：Backend 为每个 Agent 计算“有效任务集”（模板 + 覆盖 + 目标任务），带版本号，作为 Fleet 的期望状态（`fleet.state.sync`，文档类型 `TaskSet`）下发。Agent 的 `TaskManager` 与当前运行的任务对比后，启动、停止或修改参数；其中线路任务再经过控制通道同步给 拨号方。新增任务类型时协议不变。

## 3. Kind 目录：积木组合

### 3.1 积木（`MoShiro.Measurement`，只有新增积木才需要写代码）

| 类别 | 积木 | 参数示例 |
| --- | --- | --- |
| Transport | `udp`、`tcp`、`tls`、`icmp` | `port`、`sni`、`nodelay`、`sndbuf` |
| Pacer（发送节奏） | `fixed`、`poisson`、`burst`、`ladder` | `rate_hz`、`mbps`、`burst_len`、`steps[]` |
| Payload（负载形态） | `fixed`、`uniform`、`choice` | `size`、`min`/`max`、`sizes[]` |
| Behavior（对端行为） | `echo`、`sink`、`source`、`connect`、`hold` | `direction`（`fwd`/`rev`/`bidir`）、`streams` |
| Sampler（采样器，可挂到任意连接上） | `tcp_info`、`socket_counters` | `every_ms` |
| Recorder（记录器） | 根据 Kind 声明的列自动生成的列式记录器 | — |

两种角色互相独立：连接角色（拨号方 永远是拨号方，适配 NAT）与数据角色（发送端、接收端、回显端，由 `direction` 决定）无关。每种积木只实现一次，Agent 共用。

### 3.2 Kind 声明（`MoShiro.Protocol/Catalog/Kinds`）

```csharp
public static readonly KindSpec TcpEcho = new(
    Id: "tcp_echo",
    Scopes: Scope.Line,
    Transport: Blocks.Tcp(nodelay: true),
    Pacer: Blocks.Fixed(rateHz: P.Int("rate_hz", @default: 20, min: 1, max: 200)),
    Payload: Blocks.Uniform(P.Int("min", 32, 8, 1400), P.Int("max", 64, 8, 1400)),
    Behavior: Blocks.Echo(),
    Samplers: [Blocks.TcpInfo(everyMs: P.Int("tcp_info_ms", @default: 100, min: 10, max: 1000))],
    Records:
    [
        Rec.Stream("msg", Col.U32("seq"), Col.U16("size"),
                   Col.Ts拨号方("t1"), Col.TsAgent("t2"), Col.TsAgent("t3"), Col.Ts拨号方("t4")),
        Rec.Stream("tcp_info", Col.TsLocal("ts"), Col.U8("side"), Col.U32("srtt_us"),
                   Col.U32("rttvar_us"), Col.U32("cwnd"), Col.U32("retrans_total"), Col.U64("bytes_acked")),
    ],
    Budget: p => Bandwidth.Echo(p.RateHz, p.Max));   // 用于校验上限和估算带宽
```

这一条声明会被四个地方使用：

- Agent：组装积木执行任务；
- Recorder：根据列定义记录数据；
- Backend：校验参数和预算，并在 `/catalog/kinds` 中展示；
- Analysis：按列名读取数据。

### 3.3 内置 Kind

| Kind | 组合 | 默认参数 | 主要服务于 |
| --- | --- | --- | --- |
| `udp_echo` | udp + fixed + fixed + echo | 10 Hz × 64 B | 基础指标、mc、hy2 |
| `udp_train` | udp + fixed/ladder + fixed + sink | 1250 B，`rev` | hy2 |
| `tcp_echo` | tcp/tls + fixed + uniform + echo + tcp_info | 20 Hz × 32–64 B，`tcp_info` 每 100 ms | mc、reality |
| `tcp_bulk` | tcp/tls + source/sink + tcp_info | `streams` 条连接并行传输（`direction` = fwd 拨号方 发 / rev Agent 发），拨号方 每 `sample_ms` 记录每条连接的累计字节数（`bytes` 流）；两端各自只采样第一条连接的内核 tcp_info；发送端受本地 `MaxBulkMbps` 约束；只用于按需测试，默认 4 秒（够量到峰值，省流量） | reality |
| `connect` | tcp/tls + fixed + connect | 每分钟 6 次 | reality，以及目标任务 |
| `hold` | tls + hold | 每 30 s 一次心跳 | reality |
| `icmp_echo` | icmp + fixed + fixed + echo | 1 Hz × 64 B | 目标任务 |

默认的线路任务模板：`udp_echo`、`tcp_echo`、`connect`（持续运行；`connect` 每分钟 6 次，记录 TCP 建连和 TLS 握手耗时，约 0.4 kbps）。其余 Kind 可以按需开启，或者以 Plan 形式运行。每一批默认任务只种入一次，在设置里用标记记住，所以删掉的默认模板不会被重新建出来；后来新增的默认任务（`connect`）在已有部署升级后会补建一次。

### 3.4 回显的四个时间戳

```text
拨号方 单调时钟         Agent 单调时钟
  t1 发送 ─────────────> t2 收到
  t4 收到 <───────────── t3 回复
```

RTT = `(t4 − t1) − (t3 − t2)`。缺少 `t2` 表示上行丢失；有 `t2`/`t3` 但缺少 `t4` 表示下行丢失。`t2 − t1` 和 `t4 − t3` 的变化量在去除偏移和漂移后，就是每个方向各自的时延抖动。

## 4. 原始块（自描述）

```text
RawChunk
├─ header（JSON，自描述）
│   chunk_id(UUIDv7)  container_version  kind  task_id  test_id?  phase?
│   agent_id  dialer_id?  target?  params（完整快照）  port  direction
│   start_utc_ms  duration_ms
│   clock：agent 单调时钟 ↔ UTC 锚点，拨号方时钟偏移估计
│   streams: [ { name: "msg", columns: [ {name:"seq", type:"u32", enc:"delta"}, … ], count } … ]
├─ 各列数据（delta / zigzag / varint，按列定义编码）
├─ Brotli 压缩（BCL 自带）
└─ crc32c
```

- 列定义写在块头中：新增列或新增 Kind 时不需要升级 `container_version`。只有容器格式本身变化时才升级。Backend 可以通用地解码任何块，不认识的列会被保留，但会被分析层忽略。
- 切块规则：持续任务每 10 秒切一块，或者记录数达到上限（默认 50 000）时提前切块。这样采集密度提高后，单块大小仍然有界，实时显示和持久化也共用同一条链路。
- 按需测试按阶段切块，阶段内同样受记录数上限约束。
- 编解码器位于 `MoShiro.Protocol/Raw`，Agent、Backend、离线工具使用同一份代码。

## 5. 列的归属与合并

两侧各自记录完整行的流（`StreamSpec.Union`，例如 `tcp_info`）：Agent 各自采样自己的套接字（Linux `TCP_INFO`，通过 `Socket.GetRawSocketOption`，不需要 P/Invoke），每一行带 `side` 标明属于哪一侧；合并时直接拼接，不连接，Agent 一侧的行每行只交出一次。拨号方 的重传反映上行方向，Agent 的重传反映下行方向（代理场景里大流量走这个方向）。非 Linux 平台没有这类采样，对应的指标在窗口里就是没有数据。

`tcp_info` 的可空 `conn` 列标识当前记录器内的一条 TCP 连接，不包含地址。Backend 按 `(side, conn)` 分别求累计计数差，避免把重连前后或同时存活的套接字混为一个序列；时间不递增的样本被忽略，任一计数器回退时重新建立基线。旧数据没有 `conn`，仍按 `side` 分组，无法事后恢复丢失的连接归属。升级时先更新 Backend 和 Agent，再更新 拨号方；旧 Agent 合并采样时会丢弃尚不认识的列。

TCP 回显的长度头和正文合为一次写入，避免在启用 `TCP_NODELAY` 时为长度头单独发送一个小段。`tcp.retrans` 仍按内核的重传段数与发送段数求比，不由应用消息数估算。

采集层只产生原始列（`seq`、`size`、`t1…t4`、`t_send`/`t_recv`、`outcome`……），不做任何统计，也不认识上层的方案或协议；丢包率、RTT、抖动由 Backend 的分析层从这些列推算。

每个记录流（`StreamSpec`）声明：

- `Driver`：行的主导方——行由它发出的包或事件决定；
- `Key`：连接键（通常是 `seq`）；
- 每一列属于哪一侧由它的时钟决定：`Ts拨号方` 属于 拨号方，`TsAgent` 属于 Agent，其余列属于主导方。

两端各自把本侧的列记在一张按键索引的环形行表里（`RowStore`），拨号方 每 10 秒把已完成的行切成列块经控制通道交给对应的 Agent，Agent 按声明合并成完整的 RawChunk 后上传：

- `Driver = 拨号方`（回显）：行取自列块，Agent 的列按键补上，缺失就是缺失（上行丢失）；
- `Driver = Agent`（单向数据流）：行取自 Agent 的记录（按列块头里的序号范围），拨号方 的列按键补上，缺失就是缺失（丢失）；
- 没有 `Key`，或只有主导方一侧有列：原样透传。

新增一种记录流只需要写声明，不需要写合并代码。拨号方 的列在规定时间内没有送达时，Agent 带着已有的数据继续，不上传只有 Agent 一侧的块。

## 6. 代码结构（`backend/shared/MoShiro.Measurement`）

同一种采集方法的两端是一对角色写在同一个文件里：拨号方（拨号方 永远是拨号方，适配 NAT）和监听方（Agent）。两端共用包的构造和校验、时钟、行表和合并；Agent 只负责组装自己那一侧。

```text
MoShiro.Measurement/                    依赖：MoShiro.Protocol（仅 BCL，零 NuGet）
├─ Core/
│  ├─ Contracts.cs         IBlock、BlockSpec、BlockEnv、DataSession、IAdmission、IFragmentSink……
│  ├─ RowStore.cs          本侧记录的行：只含本侧拥有的列，按键索引的环形缓冲，热路径不分配
│  ├─ MonoClock.cs         单调时钟（微秒）与 UTC 锚点
│  └─ RunningBlock.cs      一个运行中的积木（取消令牌 + 后台任务）
├─ Net/
│  ├─ MsQuic*.cs、QuicDatagramTransport.cs、MeasurementDatagramRouter.cs   MsQuic 数据面与逐流认证分发
│  ├─ LinuxIcmpEcho.cs、TcpInfo.cs   ICMP 回显；Linux `TCP_INFO` 采样
│  └─ Tcp.cs               TcpLink（拨号：TLS 按指纹固定 + 认证首帧）、TcpAcceptor（监听：TLS 自动识别 + 认证首帧）
├─ Blocks/                 一个文件 = 一种采集方法的两个角色
│  ├─ BlockBase.cs         积木的公共基类
│  ├─ Echo.cs              DatagramEchoDialer/Listener、TcpEchoDialer/Listener
│  ├─ Session.cs           ConnectDialer、HoldDialer、IcmpEchoSolo（没有对端）、IdleListener
│  ├─ Train.cs             TrainDialer（接收，记 t_recv）、TrainListener（按码率发包，记发送行）
│  ├─ Bulk.cs              `tcp_bulk` 的两个角色
│  └─ BlockCatalog.cs      Kind → (拨号方, 监听方)；capabilities、本地预算的支持列表都从这里推出来
├─ Merging/                FragmentMerger（声明驱动的按键合并）、ChunkFinalizer（盖章：块头、时钟锚点、起止时间）
├─ Execution/              BudgetGuard（参数校验 + 本地硬上限，超出即拒绝）
├─ Platform/               主机汇总采样、IPv6 诊断、本机地址、平台识别
└─ Update/                 RestartUpdater（收到更新请求后有序退出）
```

边界（由架构测试强制）：采集层的任何代码里不能出现 REALITY、Hysteria2、Minecraft 这些业务概念，也没有测试方案——方案（例如 `hy2_ladder`、`reality_session`）只是若干采集原语阶段的编排，只存在于 Backend。监听方的准入策略（重放窗口、令牌桶、源地址学习）是安全边界，通过 `IAdmission` 由 Agent 实现，积木本身不含策略。

## 7. 资源与安全上限

| 项目 | 默认值 | 由谁约束 |
| --- | --- | --- |
| 同时运行的重负载测试 | 每个 Agent 1 个；拨号方 全局 1 个 | Backend 编排 + 节点本地 |
| 单次测试时长 / 码率 | ≤ 120 s；≤ 本地 `max_total_mbps` | 节点本地硬上限 |
| 单块记录数 | ≤ 50 000 | Recorder |
| 拨号方 上未交付的列数据 | 每个 Agent ≤ 8 MB，超出则丢弃最旧的数据并计入缺口 | 拨号方 |
| 持续任务的总带宽 | 每条线路 ≤ 100 kbps（默认模板约 40 kbps） | Backend 校验 |
| 接收端速率校验 | 实际速率 > 声明速率 × 1.2 时中止 | 节点 |

## 8. 端口

| 用途 | 默认端口 | 说明 |
| --- | --- | --- |
| 拨号方 控制通道（TLS，Fleet 点对点会话） | 7443/tcp | |
| QUIC Datagram 数据（`udp_echo`、`udp_train`） | 7443/udp | Agent 固定监听；拨号方 每次新连接轮换本地源端口 |
| TCP/TLS 数据（`tcp_echo`、`tcp_bulk`、`connect`、`hold`） | 7444/tcp | 按首帧中的 `stream_id` 分发 |

- 默认不使用 443，避免与 VPS 上已有的 REALITY 或 Hysteria2 服务冲突，也不需要特权端口权限。
- 任务参数中的 `port` 只能取 Agent 本地配置 `listen` 列表中的端口。要用 443 测量，需要节点所有者在本机把 443 加入 `listen`（并授予 `CAP_NET_BIND_SERVICE`），然后在任务中指定 `port: 443`。
- 每个原始块的头部都记录实际使用的端口。当测量端口与被推算服务的常用端口（443）不同时，分析结果附带 `port_differs_from_service` 置信说明，因为运营商的 QoS 可能按端口区别对待。

## 9. 边界

采集端只模拟流量特征（端口、包大小、速率、握手、方向），不实现 REALITY 或 Hysteria2 协议本身，分析结果附带置信说明。用户态看不到 TCP 的逐段收发，TCP 的原始数据是逐消息时间戳加内核采样；pcap / eBPF 见[路线图](../roadmap.md)。