基于 gRPC-Web 与 ConnectRPC 重构视频会议信令交互层的工程化实践

syntax = "proto3";
package com.company.meeting.v1;

import "google/protobuf/timestamp.proto";
import "google/protobuf/empty.proto";

service SignalingService {
  // 单播信令（Offer/Answer/Candidate）
  rpc Signal (SignalRequest) returns (SignalResponse) {}
  
  // 服端主动推送（用户加入/离开、静音状态、服务端录制通知）
  rpc SubscribeEvents (SubscribeRequest) returns (stream Event) {}
  
  // 客户端上报统计/心跳（客户端流）
  rpc ReportStats (stream StatsReport) returns (google.protobuf.Empty) {}
}

// 统一信令载体，利用 oneof 实现多态
message SignalRequest {
  string meeting_id = 1;
  string from_user_id = 2;
  string to_user_id = 3; // 空表示广播
  oneof payload {
    SessionDescription sdp = 10;
    IceCandidate candidate = 11;
    ControlCommand control = 12; // 静音/开摄像头/踢人等
  }
}

// signaling-client.ts
import { createConnectTransport, createGrpcWebTransport } from '@connectrpc/connect-web';
import { SignalingService } from '@company/proto-signaling';

// 策略：现代浏览器优先 Connect，降级 gRPC-Web
const isModernBrowser = typeof ReadableStream !== 'undefined' && 
                        typeof fetch().then(r => r.body?.pipeTo) === 'function';

export const transport = isModernBrowser 
  ? createConnectTransport({ baseUrl: '/api/connect' }) 
  : createGrpcWebTransport({ baseUrl: '/api/grpcweb' });

// 统一客户端创建
export const signalingClient = createClient(SignalingService, transport);

// 使用 AsyncIterable 原生消费流，自动处理背压
for await (const event of signalingClient.subscribeEvents({ meetingId })) {
  switch (event.payload.case) {
    case 'userJoined':
      handleUserJoined(event.payload.value);
      break;
    case 'userLeft':
      handleUserLeft(event.payload.value);
      break;
    case 'mediaStateChanged':
      updateRemoteMediaState(event.payload.value);
      break;
  }
}
// 流异常关闭时，上层自动触发重连逻辑（见 3.4 节）

// 伪代码：核心重连逻辑
async function *eventStreamWithRetry(params) {
  let lastEventId: string | undefined;
  while (state !== 'Closed') {
    try {
      const stream = client.subscribeEvents({ ...params, lastEventId });
      for await (const evt of stream) {
        lastEventId = evt.eventId;
        yield evt;
      }
    } catch (e) {
      if (isRetriable(e)) {
        await sleep(backoff());
        state = 'Reconnecting';
        continue;
      }
      throw e; // 非重试错误上抛
    }
  }
}

// 统一错误拦截器
const errorInterceptor: Interceptor = (next) => async (req) => {
  try {
    return await next(req);
  } catch (err) {
    const connectErr = ConnectError.from(err);
    // 业务错误码映射：如 MEETING_NOT_EXIST -> 跳转 404 页
    // 基础设施错误：UNAVAILABLE/DEADLINE_EXCEEDED -> 触发重连/降级提示
    // 认证错误：UNAUTHENTICATED -> 触发 Token 刷新或登出
    reportErrorToSentry(connectErr, { meetingId: req.message.meetingId });
    throw connectErr; // 保持类型给上层 catch
  }
};

# Envoy 动态生成的 Listener 片段 (LDS)
listener:
  name: signaling_ingress_443
  address: { socket_address: { address: 0.0.0.0, port_value: 443 } }
  filter_chains:
  # 1. 优先匹配 Connect 协议 (application/connect+proto, application/connect+json)
  - filter_chain_match:
      transport_protocol: "raw_buffer" # 或 tls
      application_protocols: ["h2", "http/1.1"]
      # 通过 Header 匹配分流
      header_matchers:
      - name: ":content-type"
        prefix_match: "application/connect"
    filters:
    - name: envoy.filters.network.http_connection_manager
      typed_config:
        "@type": type.googleapis.com/envoy.extensions.filters.network.http_connection_manager.v3.HttpConnectionManager
        stat_prefix: ingress_connect
        route_config:
          name: connect_route
          virtual_hosts:
          - name: signaling_service
            domains: ["*"]
            routes:
            - match: { prefix: "/com.company.meeting.v1.SignalingService/" }
              route: 
                cluster: signaling_connect_upstream # 直连支持 Connect 协议的 Go 服务
                timeout: 0s # 流式 RPC 禁用超时，依赖应用层心跳
                idle_timeout: 3600s
        http_filters:
        - name: envoy.filters.http.router
  # 2. 匹配 gRPC-Web (application/grpc-web, application/grpc-web+proto)
  - filter_chain_match:
      header_matchers:
      - name: ":content-type"
        prefix_match: "application/grpc-web"
    filters:
    - name: envoy.filters.network.http_connection_manager
      typed_config:
        "@type": type.googleapis.com/envoy.extensions.filters.network.http_connection_manager.v3.HttpConnectionManager
        stat_prefix: ingress_grpcweb
        # 关键：启用 grpc_web 转码器
        http_filters:
        - name: envoy.filters.http.grpc_web
          typed_config:
            "@type": type.googleapis.com/envoy.extensions.filters.http.grpc_web.v3.GrpcWeb
        - name: envoy.filters.http.router
        route_config:
          name: grpcweb_route
          virtual_hosts:
          - name: signaling_service
            domains: ["*"]
            routes:
            - match: { prefix: "/" }
              route:
                cluster: signaling_grpc_upstream # 后端统一 gRPC 服务
                upgrade_configs:
                - upgrade_type: CONNECT # 支持 CONNECT 方法用于 WebSocket 兼容

# CDS 资源片段
cluster:
  name: signaling_grpc_upstream
  type: EDS
  lb_policy: LEAST_REQUEST # 最少请求负载均衡，配合慢启动
  health_checks:
  - timeout: 2s
    interval: 10s
    unhealthy_threshold: 3
    healthy_threshold: 2
    grpc_health_check: { service_name: "com.company.meeting.v1.SignalingService" }
  circuit_breakers:
    thresholds:
    - max_connections: 100000      # 单实例最大连接数
      max_pending_requests: 5000   # 排队上限
      max_requests: 20000          # 并发请求上限
      max_retries: 3               # 重试预算
      track_remaining: true
      # 连接池熔断触发指标暴露给 Prometheus: envoy_cluster_upstream_cx_overflow
  outlier_detection:
    consecutive_5xx: 3
    interval: 10s
    base_ejection_time: 30s
    max_ejection_percent: 50       # 最大摘除 50% 节点，防止雪崩
    min_health_percent: 30         # 至少保留 30% 健康节点

// internal/signaling/session.go
package signaling

import (
    "context"
    "sync"
    "time"

    "connectrpc.com/connect"
    meetingv1 "github.com/company/proto/gen/go/meeting/v1"
    "go.uber.org/zap"
)

// Session 代表一个会议实例的服端状态
type Session struct {
    mu           sync.RWMutex
    meetingID    string
    participants map[string]*Participant // userID -> *Participant
    events       *EventBroker            // 基于 Ring Buffer 的广播器
    createdAt    time.Time
    logger       *zap.Logger
}

type Participant struct {
    UserID       string
    DeviceID     string
    JoinTime     time.Time
    // 三条流的发送端控制器
    SignalStream   *connect.ServerStream[meetingv1.SignalResponse] // Unary 复用为双向流模拟或单独处理
    EventStream    *connect.ServerStream[meetingv1.Event]          // Server Streaming
    StatsStream    *connect.ClientStream[meetingv1.StatsReport]    // Client Streaming (接收端)
    
    // 状态机
    State        ParticipantState // Joined, Leaving, Kicked
    LastHeartbeat time.Time
}

// EventBroker 无锁广播核心 (简化版 Ring Buffer)
type EventBroker struct {
    mu       sync.Mutex
    subs     map[chan *meetingv1.Event]struct{} // 订阅者通道集合
    buffer   []*meetingv1.Event                 // 最近 N 条事件，用于晚加入/重连回放
    maxBuf   int
}

func (s *Server) SubscribeEvents(
    ctx context.Context,
    req *connect.Request[meetingv1.SubscribeRequest],
    stream *connect.ServerStream[meetingv1.Event],
) error {
    meetingID := req.Msg.MeetingId
    userID := getUserIDFromContext(ctx) // 从 Interceptor 注入
    
    // 1. 校验会议/权限
    session := s.registry.GetOrCreate(meetingID)
    participant := session.RegisterParticipant(userID, stream)
    defer session.UnregisterParticipant(userID)

    // 2. 发送历史事件 (支持 last_event_id 断点续传)
    if req.Msg.LastEventId != "" {
        if err := session.ReplayEventsSince(req.Msg.LastEventId, stream); err != nil {
            return connect.NewError(connect.CodeInternal, err)
        }
    }

    // 3. 主循环：阻塞等待上下文取消 或 发送缓冲区满
    // Connect 的 ServerStream.Send() 内部已处理 HTTP/2 流控窗口 (WINDOW_UPDATE)
    // 我们只需监听 ctx.Done() 实现优雅关闭
    
    // 启动心跳检测协程 (服端主动 Ping，客户端无需发 Ping 帧)
    ticker := time.NewTicker(15 * time.Second)
    defer ticker.Stop()

    for {
        select {
        case <-ctx.Done():
            s.logger.Info("Event stream closed by client", zap.String("user", userID), zap.Error(ctx.Err()))
            return connect.NewError(connect.CodeCanceled, ctx.Err())
        case <-ticker.C:
            // 发送协议层心跳 (业务层 Heartbeat Event)，保持 HTTP/2 连接活性
            // 避免中间代理 (Nginx/ALB) 因空闲超时切断连接
            hbEvent := &meetingv1.Event{
                EventId:   generateEventID(),
                Timestamp: timestamppb.Now(),
                Payload: &meetingv1.Event_Heartbeat{Heartbeat: &meetingv1.Heartbeat{ServerTime: timestamppb.Now()}},
            }
            if err := stream.Send(hbEvent); err != nil {
                return connect.NewError(connect.CodeUnavailable, err) // 连接已断
            }
        }
    }
}

func (s *Server) ReportStats(
    ctx context.Context,
    stream *connect.ClientStream[meetingv1.StatsReport],
) error {
    meetingID := getMeetingIDFromContext(ctx)
    userID := getUserIDFromContext(ctx)
    
    // 批量聚合缓冲区，减少下游时序库写入压力
    batch := make([]*meetingv1.StatsReport, 0, 50)
    ticker := time.NewTicker(10 * time.Second) // 10s 刷盘一次
    defer ticker.Stop()

    for {
        select {
        case <-ctx.Done():
            s.flushStatsBatch(meetingID, userID, batch) // 退出前刷盘
            return ctx.Err()
        case <-ticker.C:
            s.flushStatsBatch(meetingID, userID, batch)
            batch = batch[:0]
        default:
            // 非阻塞接收，避免 ticker 与 Recv 竞争
            msg, err := stream.Recv()
            if err != nil {
                if connect.CodeOf(err) == connect.CodeCanceled || connect.CodeOf(err) == connect.CodeUnknown {
                    return nil // 正常关闭
                }
                s.logger.Warn("Recv stats error", zap.Error(err))
                return err
            }
            // 采样/过滤：仅保留关键指标 (rtt, jitter, packet_loss, nack_count)
            if isKeyMetric(msg) {
                batch = append(batch, msg)
            }
        }
    }
}

# PrometheusRule
groups:
- name: signaling-critical
  rules:
  # 核心流断连率飙升
  - alert: SignalingStreamDropRateHigh
    expr: |
      (rate(signaling_stream_closed_total{reason=~"error|cancel|timeout"}[5m]) 
       / rate(signaling_stream_started_total[5m])) > 0.05
    for: 2m
    labels: { severity: critical, team: rtc }
    annotations:
      summary: "信令流异常断连率 > 5%"
      description: "会议 {{ $labels.meeting_id }} 近 5 分钟异常断连率 {{ $value | humanizePercentage }}，疑似网关熔断或服端 OOM。"

  # 广播延迟 P99 超阈值
  - alert: SignalingBroadcastLatencyP99High
    expr: |
      histogram_quantile(0.99, rate(signaling_broadcast_latency_seconds_bucket[5m])) > 2
    for: 5m
    labels: { severity: warning }
    annotations:
      summary: "信令广播延迟 P99 > 2s"
      description: "可能存在慢消费者阻塞广播协程，或 Redis PubSub 堆积。"

// Interceptor 示例：消息级限流
func RateLimitInterceptor(limiter *ratelimit.Limiter) connect.UnaryInterceptorFunc {
    return func(next connect.UnaryFunc) connect.UnaryFunc {
        return func(ctx context.Context, req connect.AnyRequest) (connect.AnyResponse, error) {
            userID := getUserID(ctx)
            if !limiter.Allow(userID, 1) { // 每次消耗 1 个令牌
                return nil, connect.NewError(connect.CodeResourceExhausted, 
                    fmt.Errorf("signal rate limit exceeded for user %s", userID))
            }
            return next(ctx, req)
        }
    }
}