# MAF 会商 SSE 多 Worker 部署说明

> 状态：Phase 2 (R2) — 单 Worker 已就绪，多 Worker 需补 Redis Pub/Sub
> 更新时间：2026-08-03

---

## 1. 当前架构（单 Worker）

`ConversationEventHub`（`app/maf_roundtable/conversation_event_hub.py`）是一个
**进程内** 的 `asyncio.Event` 通知器：

- 每个 worker 进程有自己独立的 hub 实例（`app.state.maf_conversation_event_hub`）
- 事件写入者（`ConversationService.send_message`）在事务提交后调用 `hub.publish()`
- SSE 订阅者通过 `hub.subscribe()` 注册一个 `asyncio.Event`，被 `publish()` 唤醒后
  从数据库查询最新事件（通知-only，不缓存事件内容）
- 心跳：15 秒超时 → `: heartbeat\n\n` SSE 注释帧

**适用场景：** `uvicorn --workers 1` / `gunicorn -w 1`，开发环境、小规模灰度。

---

## 2. 多 Worker 限制

当使用 `--workers 2` 或更多 worker 时：

| 问题 | 原因 | 影响 |
|------|------|------|
| **跨 Worker 不通知** | Worker A 写入事件后 `hub.publish()` 只通知 A 自己的订阅者；Worker B 的 SSE 客户端收不到实时推送 | SSE 流"卡住"直到客户端超时重连，重连后从 DB 重放才能补到事件 |
| **无跨进程共享状态** | `asyncio.Event` 是进程内对象，不能跨进程通信 | 需要 Redis Pub/Sub 或等价中间件 |

**在接入 Redis Pub/Sub 之前，禁止多 Worker 灰度上线。**

---

## 3. 健康检查建议

建议在 `GET /api/maf-roundtable/health` 中增加：

```json
{
  "sse_hub_type": "in_memory",
  "sse_multi_worker": false,
  "sse_warning": "Single-worker only. Multi-worker requires Redis Pub/Sub."
}
```

当前版本已通过 `sse_multi_worker: false` 明确标注不支持多 Worker。
部署脚本应检查此项，在多 Worker 模式下发出告警。

---

## 4. Redis Pub/Sub 接入方案（后续阶段）

当需要多 Worker 时，替换 `ConversationEventHub` 为 Redis-backed 实现：

### 4.1 接口不变

```python
class ConversationEventHub(abc.ABC):
    async def publish(self, conversation_id: str) -> None: ...
    async def subscribe(self, conversation_id: str, ...) -> AsyncIterator[bool]: ...
```

### 4.2 Redis 实现

```python
class RedisConversationEventHub(ConversationEventHub):
    def __init__(self, redis: Redis, *, heartbeat_interval: float = 15.0):
        self._redis = redis
        self._heartbeat_interval = heartbeat_interval

    async def publish(self, conversation_id: str) -> None:
        await self._redis.publish(f"maf:conv:{conversation_id}", "1")

    async def subscribe(self, conversation_id: str, ...):
        pubsub = self._redis.pubsub()
        await pubsub.subscribe(f"maf:conv:{conversation_id}")
        try:
            while True:
                msg = await asyncio.wait_for(
                    pubsub.get_message(ignore_subscribe_messages=True, timeout=...),
                    timeout=self._heartbeat_interval,
                )
                if msg is None:
                    yield False  # heartbeat
                else:
                    yield True  # new events might be available
        finally:
            await pubsub.unsubscribe(f"maf:conv:{conversation_id}")
```

### 4.3 选择逻辑

```python
# deps.py
if config.database and config.database.redis_url:
    hub = RedisConversationEventHub(create_redis(config.database.redis_url))
else:
    hub = ConversationEventHub()  # in-memory, single-worker
```

---

## 5. 部署检查清单

- [ ] 确认 `uvicorn --workers 1`（或等价单进程配置）
- [ ] 确认 SSE 端点 `GET /api/maf-roundtable/conversations/{id}/stream` 可访问
- [ ] 确认 nginx / 反向代理未缓冲 SSE（`X-Accel-Buffering: no` 生效）
- [ ] 确认心跳间隔 15s 足够长不会误触发代理超时（proxy_read_timeout 应 ≥ 60s）
- [ ] 确认 `after_seq` 重连不产生重复气泡（前端 reducer 按 seq 去重）
- [ ] 多 Worker 上线前：接入 Redis Pub/Sub 并更新 `sse_multi_worker: true`