MQTT over WebSocket 多段序列发送时末帧 FIN 恒为 false

### 组件

WebSocket

### 版本号

4.3.8

### .NET SDK

.NET8

### 项目类型

WebApi

### 操作系统

Windows 11

### 运行环境

开发环境 (Development)

### 这个问题是否已经存在？

- [x] 我已经搜索过现有的问题 (https://gitee.com/RRQM_Home/TouchSocket/issues)

### 如何复现

## 复现

完整可运行工程（纯官方 NuGet 包，未做任何改动；已逐字节校验与 `4.3.8` 官方 DLL 一致）：

**https://gitee.com/LFDCC/TouchSocket/tree/demo/ws-fin-repro**

```bash
git clone -b demo/ws-fin-repro https://gitee.com/LFDCC/TouchSocket.git ts-ws-fin-repro
cd ts-ws-fin-repro
dotnet run -c Release
```

> 该分支只含复现工程（`Program.cs` / `Repro.csproj` / `README.md`），不含对 TouchSocket 的任何改动。

程序分两部分：

**第一部分**（无网络依赖，最直接）：用库自己的 `MqttPublishMessage.Build()` + `SegmentedBytesWriter` 构造报文，对同一序列分别用"当前写法"和"枚举器预读"打印每段的判定结果。

```
负载 3072 字节（报文跨段）
   报文编码长度 = 3087 字节，分 2 段，IsSingleSegment = False
   [A 当前写法] 第1段 position=(BufferSegmentNode,0)  Equals(End)=False  应为末段=False
   [A 当前写法] 第2段 position=(null,0)  Equals(End)=False  应为末段=True
   [B 枚举器预读] 第1段 hasNext=True   FIN=False  应为末段=False
   [B 枚举器预读] 第2段 hasNext=False  FIN=True   应为末段=True
   → [A] 带 FIN 的帧数 = 0  ❌ FIN 永远置不上，客户端永久挂起
```

**第二部分**（端到端）：起一个最小 Broker（TCP + WS 共享同一 `MqttBroker`），一个 WS 订阅者，服务端逐尺寸发布，打印每个尺寸是否收到。输出即上表。

关键代码：

```csharp
var received = new List<int>();
var wsClient = new MqttWebSocketClient();
await wsClient.SetupAsync(new TouchSocketConfig()
    .SetRemoteIPHost($"ws://127.0.0.1:{WsPort}/mqtt")
    .SetMqttConnectOptions(o => { o.ClientId = "repro-ws-subscriber"; o.Version = MqttProtocolVersion.V311; o.CleanSession = true; })
    .ConfigurePlugins(a => a.Add(new ReceivedCounter(received.Add))));
await wsClient.ConnectAsync(default);
await wsClient.SubscribeAsync(new MqttSubscribeMessage(new SubscribeRequest(Topic, QosLevel.AtMostOnce)), default);

// 服务端发布
await broker.ForwardMessageAsync(new MqttArrivedMessage(Topic, QosLevel.AtMostOnce, false, payload));
```

> 注：复现里 TCP/WS 两条接入共用一个 `MqttBroker`（默认实现两侧各自 `new`，客户端互不可见），为此提供了 `SharedTcpService` / `SharedWebSocketService` 两个薄封装，与 `VG.BrokerCore` 中的做法一致。若只用 WS 单侧接入，可直接用 `MqttTcpService` 的默认实现简化。

### 预期结果

MQTT over WebSocket 订阅者应能完整收到服务端发布的每一条报文，无论负载大小。
按 RFC 6455，服务端发完一条消息时，最后一个 WebSocket 帧应带 `FIN=1`，客户端据此交付应用层。

复现程序中发布三条可区分的消息后，客户端应用层应收到：

```
#1  长度= 8000  首=AAA  尾=ZZZ     ← 负载 8000 字节（编码后跨段）
#2  长度=  100  首=BBB  尾=YYY
#3  长度=  100  首=CCC  尾=XXX
```

且每条消息应在发布后立即到达（延迟应在毫秒级），不应依赖"下一条消息何时到来"。

---

### 实际结果

**负载 ≥ 3072 字节的报文不会立即送达，而是滞留在客户端，直到下一条消息到来才与之一并交付。**

实测输出（`demo/ws-fin-repro` 分支的 `dotnet run -c Release`）：

```
发布 [A]（8000 字节）后（等 800ms），应用层累计收到 0 条
  再静默等待 5 秒（期间不发送任何消息）…
  静默 5 秒后，应用层累计收到 0 条  ← 大报文永久滞留，客户端永远等不到

发布 [B]（100 字节）后，应用层累计收到 2 条      ← B 到达时，把滞留的 A 一起带出来了
发布 [C]（100 字节）后，应用层累计收到 3 条
```

**注意两次交付的内容**（应用层最终确实拿到了 A、B、C 三条，内容正确）：

```
#1  长度= 8000  首=AAA  尾=ZZZ
#2  长度=  100  首=BBB  尾=YYY
#3  长度=  100  首=CCC  尾=XXX
```

因此问题的准确表述是：**不是"数据丢失"，而是"交付被延迟到下一条消息"**——这正是它难以排查的原因：

1. **静默环境下报文永不送达**：若该报文之后长时间没有新数据（如低频状态上报、告警、控制指令），订阅者永远收不到。实测静默 5 秒后仍是 0 条。
2. **高频率下表现为"延迟一拍"**：数据持续流动时，每条消息都要等下一帧到达才被交付，订阅者看到的**永远是上一时刻的数据**。若业务对时序敏感或数据本身是"变化量"而非"全量快照"，会造成状态错乱。
3. **不抛任何异常、无任何日志**：连接正常、心跳正常、TCP 层无错误，从日志完全看不出问题。

### 分界点精确为 3072 字节（不是 4096）

| 负载 | 编码后报文 | 段数 | 结果 |
|---|---|---|---|
| 3000 B | 3015 B | 1 | ✅ 立即送达 |
| **3071 B** | 3086 B | 1 | ✅ 立即送达 |
| **3072 B** | 3087 B | **2** | ❌ 滞留，等下一帧 |
| 4000 B | 4015 B | 2 | ❌ 滞留 |
| 8000 B | 8015 B | 2 | ❌ 滞留 |

分界与主题名长度无关，只取决于负载大小。原因是 `MqttMessage.Build()` 的
`CreateVariableWriter` 会预取 `writer.GetMemory(Payload.Length + 1024)`，
而 `SegmentedBytesWriter` 首段 4096 字节、写完 1 字节固定头后可用 4095，
故 `Payload.Length + 1024 > 4095` 即**负载 ≥ 3072 时新增第二段**，从此刻起走多段发送路径。

### 异常信息

**没有任何异常。** 这是该缺陷最需要重视的地方——服务端与客户端均不抛异常、不记日志、
连接与心跳全部正常，只能从 WebSocket 线路层面观测到。

下面是用裸 TCP 抓取的真实帧数据（不经过任何 WS 客户端库，程序见 demo 仓库的抓帧部分）：

### 负载 3071 字节（单段，正常）

```
帧：opcode=BIN   FIN=1  长度=3086        ← 单帧且 FIN=1，消息完整
```

### 负载 3072 字节（跨段，异常）

```
帧：opcode=BIN   FIN=0  长度=1           ← FIN=0，消息未结束
帧：opcode=CONT  FIN=0  长度=3086        ← 仍是 FIN=0
→ 数据帧 2 个，其中带 FIN 的 0 个  ❌ 整个消息永不结束
```

### 负载 8000 字节（跨段，异常）

```
帧：opcode=BIN   FIN=0  长度=1
帧：opcode=CONT  FIN=0  长度=8014
→ 数据帧 2 个，其中带 FIN 的 0 个  ❌ 整个消息永不结束
```

### 连续两条大报文：分片状态机失控

```
负载 3072 字节（第 1 条）
   帧：opcode=BIN   FIN=0  长度=1
   帧：opcode=CONT  FIN=0  长度=3086

负载 3072 字节（第 2 条）
   帧：opcode=CONT  FIN=0  长度=1        ← 首帧变成 CONT 而非 BIN
   帧：opcode=CONT  FIN=0  长度=3086
```

第 2 条消息的首帧 opcode 变成 `CONT` 而不是 `BIN`——因为在库的分片状态机眼里，
上一条消息从未结束（`m_isCont` 未被复位），两条消息在协议层被连成了一条。

### 关键证据：应用层交付内容

```
发布 [A]（8000 字节）后，静默 5 秒 → 收到 0 条
发布 [B]（100 字节）后           → 累计 2 条（A 和 B 一起被交付）
```

即 `FIN=0` 让客户端一直把后续帧当作**同一条消息的延续**累积，直到某个带 `FIN=1` 的帧出现才一次性交付。

---

## 根因与修复建议（供作者参考）

### 根因

`src/TouchSocket.Http/WebSockets/Extensions/WebSocketClientExtension.cs`

```csharp
public static async Task SendAsync(this IWebSocket webSocket, ReadOnlySequence<byte> sequence,
    WSDataType dataType, CancellationToken cancellationToken = default)
{
    ...
    var isFirst = true;
    var position = sequence.Start;

while (sequence.TryGet(ref position, out var memory))
    {
        var isLast = position.Equals(sequence.End);   // ← 恒为 false
        var frame = new WSDataFrame(memory)
        {
            FIN = isLast,                              // ← 永远是 0
            Opcode = isFirst ? dataType : WSDataType.Cont
        };
        await webSocket.SendAsync(frame, isLast, cancellationToken).ConfigureDefaultAwait();
        isFirst = false;
    }
}
```

`position.Equals(sequence.End)` 恒为 `false`，因此多段序列的每一帧 `FIN` 都是 `false`。

`SequencePosition` 是 `readonly struct`，其 `Equals` **按字段值比较**：

```csharp
public bool Equals(SequencePosition other)
{
    if (_integer == other._integer)
        return object.Equals(_object, other._object);
    return false;
}
```

而 `TryGet` 推进出的 `position` 与 `sequence.End` **语义不同**
（.NET 8 `System.Memory.dll` 中 `ReadOnlySequence<T>.TryGetBuffer`）：

| | `_object` | `_integer` |
|---|---|---|
| 非末段推进后 | 下一节点 | **固定 0** |
| 末段推进后 | **`null`**（该分支不设 `next`） | **0** |
| `sequence.End` | 末节点 `_endObject` | 末节点写入长度 `_endInteger` |

两者字段值不可能相等。实测（`1 + 4117` 两段序列，`position` 与 `End` 的字段值）：

```
第1段 position=(BufferSegmentNode, 0)  Equals(End)=False   ← 节点是同一引用，但偏移不同（0 vs 4117）
第2段 position=(null,             0)  Equals(End)=False   ← 完全退化
```

注意第 1 段：**节点引用完全相同，仍判定为 false**，可见与"是否结构体副本"无关，纯粹是字段值不匹配。

### 修复

改用**枚举器预读**判定末帧。库内 `MqttWebSocketClient.SendAsync`（客户端侧）用的正是这种写法：

```csharp
var isFirst = true;
var enumerator = sequence.GetEnumerator();
var hasNext = enumerator.MoveNext();

while (hasNext)
{
    var memory = enumerator.Current;
    hasNext = enumerator.MoveNext();

var frame = new WSDataFrame(memory)
    {
        FIN = !hasNext,
        Opcode = isFirst ? dataType : WSDataType.Cont
    };

await webSocket.SendAsync(frame, !hasNext, cancellationToken).ConfigureDefaultAwait();
    isFirst = false;
}
```

等价判据是 `position.GetObject() is null`（推进后）——序列枚举器的 `MoveNext`
内部就是这么判终止的（`if (_next.GetObject() == null) return false;`），语义一致。

### 注意：`Slice(position).IsEmpty` 不可用作替代判据

```csharp
public ReadOnlySequence<T> Slice(SequencePosition start)
{
    bool flag = start.GetObject() != null;
    BoundsCheck(in start, flag);
    return SliceImpl(flag ? start : Start);   // ← GetObject()==null 时回退成 Start
}
```

末段推进后 `position` 退化为 `default(null, 0)`，`GetObject()` 为 null →
**回退成 `Slice(Start)`**，返回**整个序列**，`IsEmpty` 恒为 `false`。实测：

```
第1段 len=    1  Slice(pos).Length= 4117  IsEmpty=False   （真实剩余 4117）
第2段 len= 4117  Slice(pos).Length= 4118  IsEmpty=False   ← 真实剩余 0，却返回整个序列
```

---

### Demo 地址（仅限Git地址）

https://gitee.com/LFDCC/TouchSocket/tree/demo%2Fws-fin-repro/

### 承诺支持

- [x] 我确定我已经对TouchSocket项目进行了“Star”操作。

### 承诺规范

- [x] 我确定已完整阅读[Issue提问规范](https://touchsocket.net/docs/current/troubleshootissue)，并按照要求填写。

### 承诺友好

- [x] 我承诺将本着相互尊重、理解和友善的态度进行交流，共同维护好 TouchSocket 来之不易的良好的社区氛围。

GVP 若汝棋茗 / TouchSocket

内容风险标识

评论 (1)

GVP若汝棋茗 / TouchSocket

内容风险标识