higress在SSE的场景下，能否支持断连续传 || Can higress support discontinuous transmission in SSE scenarios

[undertaker86001]

需求来源: 社区用户

需求描述: higress在SSE的场景下，能否支持断连续传？ 举个例子，大模型服务通过higress回传事件流的时候，客户端因为网络问题和网关断开连接了，过一段时间后客户端又和网关建立了连接，这个时候网关能否从上次断开连接的offset继续回传事件流？

当前 Higress 在 SSE 场景下的断连续传能力

答案：目前 Higress 不支持断连续传功能

Higress 目前在 SSE 场景下不支持从上次断开连接的 offset 继续回传事件流。

SSE (Server-Sent Events) 断连续传方案中的 Last-Event-ID

Last-Event-ID 的含义和作用

Last-Event-ID 是 SSE 标准中定义的一个 HTTP 请求头，用于实现断连续传功能。工作原理如下：

1. SSE 事件结构

在 SSE 协议中，每个事件都可以包含一个唯一的 ID。 Higress 的代码中定义了 StreamEvent 结构体，其中包含了 Id 字段。

2. 事件 ID 的设置

在 SSE 流中，事件的格式通常是这样的：

id: 12345
event: message
data: {"content": "Hello World"}

从代码中可以看到， model.go:407-420的 SetValue 方法处理了事件 ID 的解析，当遇到 id: 字段时会设置到 StreamEvent.Id 中。

3. Last-Event-ID 的工作机制

当客户端与服务器的 SSE 连接断开后，客户端重新连接时会在 HTTP 请求头中携带 Last-Event-ID，其值为最后成功接收到的事件的 ID。服务器收到这个头部后，应该从该 ID 之后的事件开始重新发送。

4. 当前 Higress 的限制

虽然 Higress 的 provider.go:710-785中 ExtractStreamingEvents 函数能够解析 SSE 事件并提取 ID，但是：

1. 无状态处理：main.go:237-289 的 onStreamingResponseBody 函数是无状态的，不会检查或处理 Last-Event-ID 头部

2. 缺乏持久化：当前实现没有将事件持久化存储，因此无法根据 Last-Event-ID 进行事件重放

5. 实现 Last-Event-ID 支持的关键点

要在 Higress 中实现 Last-Event-ID 支持，需要：

1. 请求头检查：在 main.go:74-142 onHttpRequestHeader 函数中添加对 Last-Event-ID 头部的检查

2. 事件持久化：扩展现有的缓存机制，将 SSE 事件按 ID 顺序存储

3. 断点续传逻辑：根据 Last-Event-ID 的值，从存储中查找对应位置并开始重放后续事件

总结

Last-Event-ID 是 SSE 标准中实现断连续传的核心机制，它允许客户端告诉服务器从哪个事件开始继续发送。虽然 Higress 当前的代码结构已经支持事件 ID 的解析，但还需要额外的持久化和状态管理机制才能完整实现这个功能。

1. 当前实现的局限性

1. 无状态处理：AI Gateway 的流式响应处理逻辑是无状态的，只处理当前接收到的数据块，没有维护客户端连接状态

2. 缺乏 Last-Event-ID 支持：目前没有 Last-Event-ID 相关的实现

3. 无事件持久化机制：虽然 StreamEvent 结构体定义了 Id 字段，但实际上没有将事件流持久化到存储中，也没有跟踪客户端的消费位置.

4. 实时转发模式：当前的 ExtractStreamingEvents 函数只负责解析 SSE 事件并实时转发，没有状态保存机制.

2. 详细实现方案

当前 Higress 的 SSE 实现主要面向实时流式传输，缺乏断连续传所需的状态保持和事件持久化机制。要实现该功能，需要对现有的 ai-proxy 和 ai-cache 插件进行较大幅度的扩展，引入事件持久化、状态管理和断点续传逻辑。需要在性能、存储成本和功能完整性之间做出权衡。

为了实现 SSE 断连续传功能，可以尝试如下实现：

2.1 架构设计

graph TB
    subgraph "客户端层"
        Client["AI 客户端"]
        LastEventID["Last-Event-ID 头部"]
    end
    
    subgraph "Higress 网关层"
        EventBuffer["事件缓冲服务"]
        SSEHandler["SSE 处理器"]
        StateManager["连接状态管理"]
    end
    
    subgraph "存储层"
        Redis["Redis 事件存储"]
        EventLog["事件日志"]
    end
    
    subgraph "AI 服务层"
        AIService["大模型服务"]
        StreamResponse["流式响应"]
    end
    
    Client --> SSEHandler
    Client --> LastEventID
    SSEHandler --> EventBuffer
    SSEHandler --> StateManager
    EventBuffer --> Redis
    StateManager --> Redis
    AIService --> StreamResponse
    StreamResponse --> EventBuffer

Loading

2.2 核心组件实现

1. 事件缓冲和持久化服务

需要扩展现有的 ai-cache 插件，添加 SSE 事件缓冲功能：

• 为每个请求会话生成唯一的 session_id
• 将所有 SSE 事件按顺序存储到 Redis 中，包含事件ID、数据、时间戳
• 实现事件过期机制，避免无限制存储

2. Last-Event-ID 处理机制

扩展 ai-proxy 插件的请求处理逻辑：

• 在 onHttpRequestHeader 中检查 Last-Event-ID 头部 4
• 如果存在该头部，从 Redis 中查找对应的事件位置
• 实现断点续传逻辑，从指定位置开始发送事件

3. 连接状态管理

• 跟踪每个客户端连接的消费位置
• 在连接断开时保存状态
• 在重连时恢复状态

2.3 实现步骤

步骤1：扩展 StreamEvent 结构

需要修改现有的事件结构，添加更多元数据：

type PersistentStreamEvent struct {
    Id        string    `json:"id"`
    Event     string    `json:"event"`
    Data      string    `json:"data"`
    Timestamp int64     `json:"timestamp"`
    SessionId string    `json:"session_id"`
    Sequence  int64     `json:"sequence"`
}

步骤2：实现事件持久化层

基于现有的 ai-cache 插件实现 5 ，创建事件持久化服务：

• 扩展 Redis 操作接口
• 实现事件的有序存储和检索
• 添加事件过期和清理机制

步骤3：修改流式响应处理

扩展现有的 onStreamingResponseBody 函数：

• 在处理每个事件时，同时写入持久化存储
• 生成单调递增的事件ID
• 维护会话状态

步骤4：实现断点续传逻辑

• 解析 Last-Event-ID 头部
• 从存储中查找对应的事件位置
• 实现从指定位置开始的事件重放

2.4 配置示例

apiVersion: networking.higress.io/v1
kind: WasmPlugin
metadata:
  name: ai-proxy-with-resume
spec:
  config:
    provider:
      type: qwen
      apiTokens:
        - "your-api-token"
    resumeSupport:
      enabled: true
      redis:
        host: "redis-server:6379"
        db: 0
      eventTTL: 3600  # 事件保存1小时
      maxEvents: 10000  # 每个会话最大事件数

3. 技术考虑

1. 性能影响：每个事件都需要额外的存储操作，会对性能产生影响
2. 存储成本：长时间的事件存储会增加 Redis 的存储成本
3. 内存管理：需要合理的事件清理策略，避免内存泄漏
4. 一致性保证：需要确保事件的顺序性和完整性

---

Source of demand: Community users

Requirement Description: Can higress support discontinuous transmission in SSE scenarios? For example, when a big model service returns event stream through higress, the client disconnects from the gateway due to network problems. After a period of time, the client establishes a connection with the gateway. At this time, can the gateway continue to return event stream from the last disconnected offset?

Current Higress's continuous transmission capability in SSE scenarios

Answer: Currently, Higress does not support the interrupted continuous transmission function

Higress currently does not support *********************************************************************************************************************************************************************************************************************************************************

SSE (Server-Sent Events) Last-Event-ID in the interrupted continuous transmission scheme

The meaning and function of Last-Event-ID

Last-Event-ID is an HTTP request header defined in the SSE standard, used to implement the interrupted continuous transmission function. The working principle is as follows:

1. SSE event structure

In the SSE protocol, each event can contain a unique ID. The StreamEvent structure is defined in the code of Higress, which contains the Id field.

2. Event ID settings

In an SSE stream, the format of events is usually like this:

id: 12345
event: message
data: {"content": "Hello World"}

As you can see from the code, the SetValue method of model.go:407-420 handles the parsing of event ID, and is set to StreamEvent.Id when the id: field is encountered.

3. Last-Event-ID working mechanism

When the client disconnects the SSE connection from the server, the client reconnects, it carries Last-Event-ID in the HTTP request header, which is the ID of the last successfully received event. After the server receives this header, it should start resending from events following the ID.

4. Current restrictions on Higress

Although the ExtractStreamingEvents function in Higress' provider.go:710-785 can parse SSE events and extract IDs, however:

1. Stateless Processing: The onStreamingResponseBody function of main.go:237-289 is stateless and will not check or process the Last-Event-ID header

2. Laboring persistence: The current implementation does not persist events, so event replay is not possible based on Last-Event-ID

5. Key points for implementing Last-Event-ID support

To implement Last-Event-ID support in Higress, you need:

1. Request header check: Add a check on the Last-Event-ID header in main.go:74-142 onHttpRequestHeader function

2. Event persistence: Extend the existing cache mechanism and store SSE events in ID order

3. Breakpoint continuous transmission logic: According to the value of Last-Event-ID, find the corresponding position from the storage and start replaying subsequent events

Summarize

Last-Event-ID is the core mechanism in the SSE standard to implement continuous transmission. It allows the client to tell the server which event to continue sending. Although the current code structure of Higress already supports the parsing of event IDs, additional persistence and state management mechanisms are required to fully implement this function.

1. Current implementation limitations

1. Stateless processing: The streaming response processing logic of AI Gateway is stateless, only processing the currently received data blocks, and does not maintain the client connection status.

2. Laboring Last-Event-ID support: There is currently no Last-Event-ID related implementation

3. Event-free persistence mechanism: Although the StreamEvent structure defines the Id field, it does not actually persist the event stream to storage, nor does it track the client's consumption location.

4. Real-time forwarding mode: The current ExtractStreamingEvents function is only responsible for parsing SSE events and forwarding them in real time, and has no state saving mechanism.

2. Detailed implementation plan

Currently, Higress’s SSE implementation is mainly aimed at real-time streaming, and lacks the state holding and event persistence mechanism required for discontinuous transmission. To implement this function, it is necessary to significantly expand the existing ai-proxy and ai-cache plug-ins, introducing event persistence, state management and breakpoint continuation logic. There is a trade-off between performance, storage cost, and functional integrity.

In order to implement the SSE interrupt continuous transmission function, you can try to implement the following:

2.1 Architecture Design

graph TB
    subgraph "Client Layer"
        Client["AI Client"]
        LastEventID["Last-Event-ID Header"]
    end
    
    subgraph "Higress Gateway Layer"
        EventBuffer["Event Buffer Service"]
        SSEHandler["SSE Processor"]
        StateManager["Connection State Management"]
    end
    
    subgraph "Storage Layer"
        Redis["Redis Event Storage"]
        EventLog["Event Log"]
    end
    
    subgraph "AI Service Layer"
        AIService["Model Service"]
        StreamResponse["Stream Response"]
    end
    
    Client --> SSEHandler
    Client --> LastEventID
    SSEHandler --> EventBuffer
    SSEHandler --> StateManager
    EventBuffer --> Redis
    StateManager --> Redis
    AIService --> StreamResponse
    StreamResponse --> EventBuffer

Loading

2.2 Core Component Implementation

1. Event buffering and persistence services

You need to extend the existing ai-cache plug-in and add SSE event buffering function:

• Generate unique session_id for each request session
• Store all SSE events in Redis in order, including event ID, data, timestamp
• Implement event expiration mechanism to avoid unlimited storage

2. Last-Event-ID processing mechanism

Extending the request processing logic of the ai-proxy plugin:

• Check the Last-Event-ID header in onHttpRequestHeader 4
• If this header exists, find the corresponding event location from Redis
• Implement breakpoint continuous transmission logic, and start sending events from the specified location

3. Connection status management

• Track the consumption location of each client connection
• Save status when connection is disconnected
• Restore state when reconnecting

2.3 Implementation steps

Step 1: Extend StreamEvent Structure

You need to modify the existing event structure and add more metadata:

type PersistentStreamEvent struct {
    Id string `json:"id"`
    Event string `json:"event"`
    Data string `json:"data"`
    Timestamp int64 `json:"timestamp"`
    SessionId string `json:"session_id"`
    Sequence int64 `json:"sequence"`
}

Step 2: Implement event persistence layer

Based on the existing ai-cache plug-in implementation 5, create event persistence services:

• Extended Redis Operation Interface
• Implement orderly storage and retrieval of events
• Add event expiration and cleaning mechanisms

Step 3: Modify streaming response processing

Extend the existing onStreamingResponseBody function:

• Write to persistent storage while processing each event
• Generate monotonically increasing event ID
• Maintain session status

Step 4: Implement breakpoint continuous transmission logic

• Analysis Last-Event-ID header
• Find the corresponding event location from storage
• Implement event replay starting from the specified location

2.4 Configuration Example

apiVersion: networking.higress.io/v1
kind: WasmPlugin
metadata:
  name: ai-proxy-with-resume
spec:
  config:
    Provider:
      type: qwen
      apiTokens:
        - "your-api-token"
    resumeSupport:
      enabled: true
      redis:
        host: "redis-server:6379"
        db: 0
      eventTTL: 3600 # Event saved for 1 hour
      maxEvents: 10000 # Maximum number of events per session

3. Technical considerations

1. Performance Impact: Each event requires additional storage operations, which will have an impact on performance
2. Storage Cost: Long-term event storage will increase the storage cost of Redis
3. Memory Management: A reasonable event cleaning strategy is required to avoid memory leaks
4. Consistency Guarantee: It is necessary to ensure the order and integrity of events