50822600ad
- Added Redis Streams operations to the message bus interface and implementation. - Introduced StreamCheckpoint model to track last processed stream entry per document. - Implemented UpsertStreamCheckpoint and GetStreamCheckpoint methods in the Postgres store. - Created document_update_history table for storing update payloads for recovery and replay. - Developed update persist worker to handle Redis Stream updates and persist them to Postgres. - Enhanced Docker Compose configuration for Redis with persistence. - Updated frontend API to support fetching document state with optional share token. - Added connection stability monitoring in the Yjs document hook.
947 lines
26 KiB
Go
947 lines
26 KiB
Go
package hub
|
||
|
||
import (
|
||
"context"
|
||
"encoding/base64"
|
||
"strconv"
|
||
"sync"
|
||
"time"
|
||
|
||
"github.com/M1ngdaXie/realtime-collab/internal/messagebus"
|
||
"github.com/google/uuid"
|
||
"github.com/gorilla/websocket"
|
||
"go.uber.org/zap"
|
||
)
|
||
|
||
type Message struct {
|
||
RoomID string
|
||
Data []byte
|
||
sender *Client
|
||
}
|
||
|
||
type Client struct {
|
||
ID string
|
||
UserID *uuid.UUID // Authenticated user ID (nil for public share access)
|
||
UserName string // User's display name for presence
|
||
UserAvatar *string // User's avatar URL for presence
|
||
Permission string // User's permission level: "owner", "edit", "view"
|
||
Conn *websocket.Conn
|
||
send chan []byte
|
||
sendMu sync.Mutex
|
||
sendClosed bool
|
||
hub *Hub
|
||
roomID string
|
||
mutex sync.Mutex
|
||
unregisterOnce sync.Once
|
||
failureCount int
|
||
failureMu sync.Mutex
|
||
observedYjsIDs map[uint64]uint64 // clientID -> maxClock
|
||
idsMu sync.Mutex
|
||
}
|
||
type Room struct {
|
||
ID string
|
||
clients map[*Client]bool
|
||
mu sync.RWMutex
|
||
cancel context.CancelFunc
|
||
reconnectCount int // Track Redis reconnection attempts for debugging
|
||
}
|
||
|
||
type Hub struct {
|
||
rooms map[string]*Room
|
||
mu sync.RWMutex
|
||
Register chan *Client // Exported
|
||
Unregister chan *Client // Exported
|
||
Broadcast chan *Message // Exported
|
||
|
||
//redis pub/sub config
|
||
messagebus messagebus.MessageBus
|
||
logger *zap.Logger
|
||
serverID string
|
||
fallbackMode bool
|
||
|
||
// P0 fix: bounded worker pool for Redis Publish
|
||
publishQueue chan *Message // buffered queue consumed by fixed workers
|
||
publishDone chan struct{} // close to signal workers to exit
|
||
|
||
subscribeMu sync.Mutex
|
||
|
||
// Bounded worker pool for Redis SetAwareness
|
||
awarenessQueue chan awarenessItem
|
||
|
||
// Stream persistence worker pool (P1: Redis Streams durability)
|
||
streamQueue chan *Message // buffered queue for XADD operations
|
||
streamDone chan struct{} // close to signal stream workers to exit
|
||
}
|
||
|
||
const (
|
||
// publishWorkerCount is the number of fixed goroutines consuming from publishQueue.
|
||
// 50 workers can handle ~2000 msg/sec assuming ~25ms avg Redis RTT per publish.
|
||
publishWorkerCount = 50
|
||
|
||
// publishQueueSize is the buffer size for the publish queue channel.
|
||
publishQueueSize = 4096
|
||
|
||
// awarenessWorkerCount is the number of fixed goroutines consuming from awarenessQueue.
|
||
awarenessWorkerCount = 8
|
||
|
||
// awarenessQueueSize is the buffer size for awareness updates.
|
||
awarenessQueueSize = 4096
|
||
|
||
// streamWorkerCount is the number of fixed goroutines consuming from streamQueue.
|
||
// 50 workers match publish workers for consistent throughput.
|
||
streamWorkerCount = 50
|
||
|
||
// streamQueueSize is the buffer size for the stream persistence queue.
|
||
streamQueueSize = 4096
|
||
)
|
||
|
||
type awarenessItem struct {
|
||
roomID string
|
||
clientIDs []uint64
|
||
data []byte
|
||
}
|
||
|
||
func NewHub(messagebus messagebus.MessageBus, serverID string, logger *zap.Logger) *Hub {
|
||
h := &Hub{
|
||
rooms: make(map[string]*Room),
|
||
Register: make(chan *Client, 2048),
|
||
Unregister: make(chan *Client, 2048),
|
||
Broadcast: make(chan *Message, 4096),
|
||
// redis
|
||
messagebus: messagebus,
|
||
serverID: serverID,
|
||
logger: logger,
|
||
fallbackMode: false,
|
||
// P0 fix: bounded publish worker pool
|
||
publishQueue: make(chan *Message, publishQueueSize),
|
||
publishDone: make(chan struct{}),
|
||
// bounded awareness worker pool
|
||
awarenessQueue: make(chan awarenessItem, awarenessQueueSize),
|
||
// Stream persistence worker pool
|
||
streamQueue: make(chan *Message, streamQueueSize),
|
||
streamDone: make(chan struct{}),
|
||
}
|
||
|
||
// Start the fixed worker pool for Redis publishing
|
||
h.startPublishWorkers(publishWorkerCount)
|
||
h.startAwarenessWorkers(awarenessWorkerCount)
|
||
h.startStreamWorkers(streamWorkerCount)
|
||
|
||
return h
|
||
}
|
||
|
||
// startPublishWorkers launches n goroutines that consume from publishQueue
|
||
// and publish messages to Redis. Workers exit when publishDone is closed.
|
||
func (h *Hub) startPublishWorkers(n int) {
|
||
for i := 0; i < n; i++ {
|
||
go func(workerID int) {
|
||
for {
|
||
select {
|
||
case <-h.publishDone:
|
||
h.logger.Info("Publish worker exiting", zap.Int("worker_id", workerID))
|
||
return
|
||
case msg, ok := <-h.publishQueue:
|
||
if !ok {
|
||
return
|
||
}
|
||
ctx, cancel := context.WithTimeout(context.Background(), 3*time.Second)
|
||
|
||
err := h.messagebus.Publish(ctx, msg.RoomID, msg.Data)
|
||
|
||
cancel()
|
||
|
||
if err != nil {
|
||
h.logger.Error("Redis Publish failed", zap.Error(err))
|
||
}
|
||
}
|
||
}
|
||
}(i)
|
||
}
|
||
h.logger.Info("Publish worker pool started", zap.Int("workers", n))
|
||
}
|
||
|
||
func (h *Hub) startAwarenessWorkers(n int) {
|
||
for i := 0; i < n; i++ {
|
||
go func(workerID int) {
|
||
for {
|
||
select {
|
||
case <-h.publishDone:
|
||
h.logger.Info("Awareness worker exiting", zap.Int("worker_id", workerID))
|
||
return
|
||
case item, ok := <-h.awarenessQueue:
|
||
if !ok {
|
||
return
|
||
}
|
||
if h.fallbackMode || h.messagebus == nil {
|
||
continue
|
||
}
|
||
ctx, cancel := context.WithTimeout(context.Background(), 2*time.Second)
|
||
for _, clientID := range item.clientIDs {
|
||
if err := h.messagebus.SetAwareness(ctx, item.roomID, clientID, item.data); err != nil {
|
||
h.logger.Warn("Failed to cache awareness in Redis",
|
||
zap.Uint64("yjs_id", clientID),
|
||
zap.Error(err))
|
||
}
|
||
}
|
||
cancel()
|
||
}
|
||
}
|
||
}(i)
|
||
}
|
||
h.logger.Info("Awareness worker pool started", zap.Int("workers", n))
|
||
}
|
||
|
||
// startStreamWorkers launches n goroutines that consume from streamQueue
|
||
// and add messages to Redis Streams for durability and replay.
|
||
func (h *Hub) startStreamWorkers(n int) {
|
||
for i := 0; i < n; i++ {
|
||
go func(workerID int) {
|
||
for {
|
||
select {
|
||
case <-h.streamDone:
|
||
h.logger.Info("Stream worker exiting", zap.Int("worker_id", workerID))
|
||
return
|
||
case msg, ok := <-h.streamQueue:
|
||
if !ok {
|
||
return
|
||
}
|
||
h.addToStream(msg)
|
||
}
|
||
}
|
||
}(i)
|
||
}
|
||
h.logger.Info("Stream worker pool started", zap.Int("workers", n))
|
||
}
|
||
|
||
// encodeBase64 encodes binary data to base64 string for Redis storage
|
||
func encodeBase64(data []byte) string {
|
||
return base64.StdEncoding.EncodeToString(data)
|
||
}
|
||
|
||
// addToStream adds a message to Redis Streams for durability
|
||
func (h *Hub) addToStream(msg *Message) {
|
||
ctx, cancel := context.WithTimeout(context.Background(), 2*time.Second)
|
||
defer cancel()
|
||
|
||
streamKey := "stream:" + msg.RoomID
|
||
|
||
// Get next sequence number atomically
|
||
seqKey := "seq:" + msg.RoomID
|
||
seq, err := h.messagebus.Incr(ctx, seqKey)
|
||
if err != nil {
|
||
h.logger.Error("Failed to increment sequence",
|
||
zap.String("room_id", msg.RoomID),
|
||
zap.Error(err))
|
||
return
|
||
}
|
||
|
||
// Encode payload as base64 (binary-safe storage)
|
||
payload := encodeBase64(msg.Data)
|
||
|
||
// Extract Yjs message type from first byte as numeric string
|
||
msgType := "0"
|
||
if len(msg.Data) > 0 {
|
||
msgType = strconv.Itoa(int(msg.Data[0]))
|
||
}
|
||
|
||
// Add entry to Stream with MAXLEN trimming
|
||
values := map[string]interface{}{
|
||
"type": "update",
|
||
"server_id": h.serverID,
|
||
"yjs_payload": payload,
|
||
"msg_type": msgType,
|
||
"seq": seq,
|
||
"timestamp": time.Now().Format(time.RFC3339),
|
||
}
|
||
|
||
_, err = h.messagebus.XAdd(ctx, streamKey, 10000, true, values)
|
||
if err != nil {
|
||
h.logger.Error("Failed to add to Stream",
|
||
zap.String("stream_key", streamKey),
|
||
zap.Int64("seq", seq),
|
||
zap.Error(err))
|
||
return
|
||
}
|
||
|
||
// Mark this document as active so the persist worker only processes active streams
|
||
_ = h.messagebus.ZAdd(ctx, "active-streams", float64(time.Now().Unix()), msg.RoomID)
|
||
}
|
||
|
||
func (h *Hub) Run() {
|
||
for {
|
||
select {
|
||
case client := <-h.Register:
|
||
h.registerClient(client)
|
||
case client := <-h.Unregister:
|
||
h.unregisterClient(client)
|
||
case message := <-h.Broadcast:
|
||
h.broadcastMessage(message)
|
||
}
|
||
}
|
||
}
|
||
|
||
func (h *Hub) registerClient(client *Client) {
|
||
var room *Room
|
||
var exists bool
|
||
var needSubscribe bool
|
||
|
||
h.mu.Lock()
|
||
room, exists = h.rooms[client.roomID]
|
||
|
||
// --- 1. 初始化房间 (仅针对该服务器上的第一个人) ---
|
||
if !exists {
|
||
room = &Room{
|
||
ID: client.roomID,
|
||
clients: make(map[*Client]bool),
|
||
cancel: nil,
|
||
// lastAwareness 已被我们从结构体中物理删除或不再使用
|
||
}
|
||
h.rooms[client.roomID] = room
|
||
h.logger.Info("Created new local room instance", zap.String("room_id", client.roomID))
|
||
}
|
||
if room.cancel == nil && !h.fallbackMode && h.messagebus != nil {
|
||
needSubscribe = true
|
||
}
|
||
h.mu.Unlock()
|
||
|
||
// 开启跨服订阅（避免在 h.mu 下做网络 I/O）
|
||
if needSubscribe {
|
||
h.subscribeMu.Lock()
|
||
h.mu.RLock()
|
||
room = h.rooms[client.roomID]
|
||
alreadySubscribed := room != nil && room.cancel != nil
|
||
h.mu.RUnlock()
|
||
if !alreadySubscribed {
|
||
ctx, cancel := context.WithCancel(context.Background())
|
||
msgChan, err := h.messagebus.Subscribe(ctx, client.roomID)
|
||
if err != nil {
|
||
h.logger.Error("Redis Subscribe failed", zap.Error(err))
|
||
cancel()
|
||
} else {
|
||
h.mu.Lock()
|
||
room = h.rooms[client.roomID]
|
||
if room == nil {
|
||
h.mu.Unlock()
|
||
cancel()
|
||
_ = h.messagebus.Unsubscribe(context.Background(), client.roomID)
|
||
} else {
|
||
room.cancel = cancel
|
||
h.mu.Unlock()
|
||
go h.startRoomMessageForwarding(ctx, client.roomID, msgChan)
|
||
}
|
||
}
|
||
}
|
||
h.subscribeMu.Unlock()
|
||
}
|
||
|
||
// --- 2. 将客户端加入本地房间列表 ---
|
||
room.mu.Lock()
|
||
room.clients[client] = true
|
||
room.mu.Unlock()
|
||
|
||
h.logger.Info("Client linked to room", zap.String("client_id", client.ID))
|
||
|
||
// --- 3. 核心改进：立刻同步全量状态 ---
|
||
// 无论是不是第一个人，只要有人进来，我们就去 Redis 抓取所有人的状态发给他
|
||
// hub/hub.go 内部的 registerClient 函数
|
||
|
||
// ... 之前的代码保持不变 ...
|
||
|
||
if !h.fallbackMode && h.messagebus != nil {
|
||
go func(c *Client) {
|
||
ctx, cancel := context.WithTimeout(context.Background(), 2*time.Second)
|
||
defer cancel()
|
||
|
||
// 1. 从 Redis 抓取
|
||
awarenessMap, err := h.messagebus.GetAllAwareness(ctx, c.roomID)
|
||
if err != nil {
|
||
h.logger.Error("Redis sync failed in goroutine",
|
||
zap.String("client_id", c.ID),
|
||
zap.Error(err))
|
||
return
|
||
}
|
||
|
||
if len(awarenessMap) == 0 {
|
||
h.logger.Debug("No awareness data found in Redis for sync", zap.String("room_id", c.roomID))
|
||
return
|
||
}
|
||
|
||
h.logger.Info("Starting state delivery to joiner",
|
||
zap.String("client_id", c.ID),
|
||
zap.Int("items", len(awarenessMap)))
|
||
|
||
// 2. 逐条发送，带锁保护
|
||
sentCount := 0
|
||
for clientID, data := range awarenessMap {
|
||
c.sendMu.Lock()
|
||
// 🛑 核心防御：检查通道是否已被 unregisterClient 关闭
|
||
if c.sendClosed {
|
||
c.sendMu.Unlock()
|
||
h.logger.Warn("Sync aborted: client channel closed while sending",
|
||
zap.String("client_id", c.ID),
|
||
zap.Uint64("target_yjs_id", clientID))
|
||
return // 直接退出协程，不发了
|
||
}
|
||
|
||
select {
|
||
case c.send <- data:
|
||
sentCount++
|
||
default:
|
||
// 缓冲区满了（通常是因为网络太卡），记录一条警告
|
||
h.logger.Warn("Sync item skipped: client send buffer full",
|
||
zap.String("client_id", c.ID),
|
||
zap.Uint64("target_yjs_id", clientID))
|
||
}
|
||
c.sendMu.Unlock()
|
||
}
|
||
|
||
h.logger.Info("State sync completed successfully",
|
||
zap.String("client_id", c.ID),
|
||
zap.Int("delivered", sentCount))
|
||
}(client)
|
||
}
|
||
}
|
||
func (h *Hub) unregisterClient(client *Client) {
|
||
h.mu.Lock()
|
||
// 注意：这里不使用 defer，因为我们需要根据逻辑流手动释放锁，避免阻塞后续的 Redis 操作
|
||
|
||
room, exists := h.rooms[client.roomID]
|
||
if !exists {
|
||
h.mu.Unlock()
|
||
return
|
||
}
|
||
|
||
// 1. 局部清理：从房间内移除客户端
|
||
room.mu.Lock()
|
||
if _, ok := room.clients[client]; ok {
|
||
delete(room.clients, client)
|
||
|
||
// 安全关闭客户端的发送管道
|
||
client.sendMu.Lock()
|
||
if !client.sendClosed {
|
||
close(client.send)
|
||
client.sendClosed = true
|
||
}
|
||
client.sendMu.Unlock()
|
||
}
|
||
remainingClientsCount := len(room.clients)
|
||
room.mu.Unlock()
|
||
|
||
h.logger.Info("Unregistered client from room",
|
||
zap.String("client_id", client.ID),
|
||
zap.String("room_id", client.roomID),
|
||
zap.Int("remaining_clients", remainingClientsCount),
|
||
)
|
||
|
||
// 2. 分布式清理：删除 Redis 中的感知数据 (Awareness)
|
||
if !h.fallbackMode && h.messagebus != nil {
|
||
go func() {
|
||
// 使用带超时的 Context 执行删除
|
||
ctx, cancel := context.WithTimeout(context.Background(), 3*time.Second)
|
||
defer cancel()
|
||
|
||
client.idsMu.Lock()
|
||
// 遍历该客户端在本机观察到的所有 Yjs ID
|
||
for clientID := range client.observedYjsIDs {
|
||
err := h.messagebus.DeleteAwareness(ctx, client.roomID, clientID)
|
||
h.logger.Info("DEBUG: IDs to cleanup",
|
||
zap.String("client_id", client.ID),
|
||
zap.Any("ids", client.observedYjsIDs))
|
||
if err != nil {
|
||
h.logger.Warn("Failed to delete awareness from Redis",
|
||
zap.Uint64("yjs_id", clientID),
|
||
zap.Error(err),
|
||
)
|
||
}
|
||
}
|
||
client.idsMu.Unlock()
|
||
}()
|
||
}
|
||
|
||
// 3. 协作清理：发送"僵尸删除"消息
|
||
// 注意：无论本地是否有其他客户端，都要发布到 Redis，因为其他服务器可能有客户端
|
||
client.idsMu.Lock()
|
||
clientClocks := make(map[uint64]uint64, len(client.observedYjsIDs))
|
||
for id, clock := range client.observedYjsIDs {
|
||
clientClocks[id] = clock
|
||
}
|
||
client.idsMu.Unlock()
|
||
|
||
if len(clientClocks) > 0 {
|
||
// 构造 Yjs 协议格式的删除消息
|
||
deleteMsg := MakeYjsDeleteMessage(clientClocks)
|
||
|
||
// 本地广播：只有当本地还有其他客户端时才需要
|
||
if remainingClientsCount > 0 {
|
||
msg := &Message{
|
||
RoomID: client.roomID,
|
||
Data: deleteMsg,
|
||
sender: nil, // 系统发送
|
||
}
|
||
go func() {
|
||
select {
|
||
case h.Broadcast <- msg:
|
||
case <-time.After(500 * time.Millisecond):
|
||
h.logger.Error("Critical: Failed to broadcast cleanup message (channel blocked)")
|
||
}
|
||
}()
|
||
}
|
||
|
||
// 发布到 Redis：无论本地是否有客户端，都要通知其他服务器
|
||
if !h.fallbackMode && h.messagebus != nil {
|
||
go func(roomID string, data []byte) {
|
||
ctx, cancel := context.WithTimeout(context.Background(), 2*time.Second)
|
||
defer cancel()
|
||
if err := h.messagebus.Publish(ctx, roomID, data); err != nil {
|
||
h.logger.Error("Failed to publish delete message to Redis",
|
||
zap.String("room_id", roomID),
|
||
zap.Error(err))
|
||
} else {
|
||
h.logger.Debug("Published delete message to Redis",
|
||
zap.String("room_id", roomID),
|
||
zap.Int("yjs_ids_count", len(clientClocks)))
|
||
}
|
||
}(client.roomID, deleteMsg)
|
||
}
|
||
}
|
||
|
||
// 4. 房间清理：如果是本服务器最后一个人，清理本地资源
|
||
// 注意：不要删除整个 Redis Hash，因为其他服务器可能还有客户端
|
||
if remainingClientsCount == 0 {
|
||
h.logger.Info("Room is empty on this server, cleaning up local resources", zap.String("room_id", client.roomID))
|
||
|
||
// A. 停止转发协程
|
||
if room.cancel != nil {
|
||
room.cancel()
|
||
}
|
||
|
||
// B. 取消 Redis 订阅（但不删除 awareness hash，其他服务器可能还有客户端）
|
||
if !h.fallbackMode && h.messagebus != nil {
|
||
go func(rID string) {
|
||
ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
|
||
defer cancel()
|
||
|
||
// 只取消订阅，不清除 awareness（已在步骤2中按 Yjs ID 单独删除）
|
||
if err := h.messagebus.Unsubscribe(ctx, rID); err != nil {
|
||
h.logger.Warn("Failed to unsubscribe", zap.Error(err))
|
||
}
|
||
}(client.roomID)
|
||
}
|
||
// C. 从内存中移除
|
||
delete(h.rooms, client.roomID)
|
||
}
|
||
|
||
h.mu.Unlock() // 手动释放 Hub 锁
|
||
}
|
||
|
||
const (
|
||
writeWait = 10 * time.Second
|
||
pongWait = 60 * time.Second
|
||
pingPeriod = (pongWait * 9) / 10 // 54 seconds
|
||
maxSendFailures = 5
|
||
)
|
||
|
||
func (h *Hub) broadcastMessage(message *Message) {
|
||
h.mu.RLock()
|
||
room, exists := h.rooms[message.RoomID]
|
||
h.mu.RUnlock()
|
||
|
||
if !exists {
|
||
// 如果房间不存在，没必要继续
|
||
return
|
||
}
|
||
|
||
// 1. 处理 Awareness 缓存 (Type 1)
|
||
// if len(message.Data) > 0 && message.Data[0] == 1 {
|
||
// room.mu.Lock()
|
||
// room.lastAwareness = make([]byte, len(message.Data))
|
||
// copy(room.lastAwareness, message.Data)
|
||
// room.mu.Unlock()
|
||
// }
|
||
|
||
// 2. 本地广播
|
||
h.broadcastToLocalClients(room, message.Data, message.sender)
|
||
|
||
// 只有本地客户端发出的消息 (sender != nil) 才推送到 Redis
|
||
// P0 fix: send to bounded worker pool instead of spawning unbounded goroutines
|
||
if message.sender != nil && !h.fallbackMode && h.messagebus != nil {
|
||
// 3a. Publish to Pub/Sub (real-time cross-server broadcast)
|
||
select {
|
||
case h.publishQueue <- message:
|
||
// Successfully queued for async publish by worker pool
|
||
default:
|
||
// Queue full — drop to protect the system (same pattern as broadcastToLocalClients)
|
||
h.logger.Warn("Publish queue full, dropping Redis publish",
|
||
zap.String("room_id", message.RoomID))
|
||
}
|
||
|
||
// 3b. Add to Stream for durability (only Type 0 updates, not Type 1 awareness)
|
||
// Type 0 = Yjs sync/update messages (document changes)
|
||
// Type 1 = Yjs awareness messages (cursors, presence) - ephemeral, skip
|
||
if len(message.Data) > 0 && message.Data[0] == 0 {
|
||
select {
|
||
case h.streamQueue <- message:
|
||
// Successfully queued for async Stream add
|
||
default:
|
||
h.logger.Warn("Stream queue full, dropping durability",
|
||
zap.String("room_id", message.RoomID))
|
||
}
|
||
}
|
||
}
|
||
}
|
||
|
||
func (h *Hub) broadcastToLocalClients(room *Room, data []byte, sender *Client) {
|
||
room.mu.RLock()
|
||
defer room.mu.RUnlock()
|
||
|
||
// 2. 遍历房间内所有本地连接的客户端
|
||
for client := range room.clients {
|
||
// 3. 排除发送者（如果是从 Redis 来的消息，sender 通常为 nil）
|
||
if client != sender {
|
||
select {
|
||
case client.send <- data:
|
||
// 发送成功：重置该客户端的失败计数
|
||
client.failureMu.Lock()
|
||
client.failureCount = 0
|
||
client.failureMu.Unlock()
|
||
|
||
default:
|
||
client.handleSendFailure()
|
||
}
|
||
}
|
||
}
|
||
}
|
||
func (h *Hub) startRoomMessageForwarding(ctx context.Context, roomID string, msgChan <-chan []byte) {
|
||
// Increment and log reconnection count for debugging
|
||
h.mu.RLock()
|
||
room, exists := h.rooms[roomID]
|
||
h.mu.RUnlock()
|
||
|
||
if exists {
|
||
room.mu.Lock()
|
||
room.reconnectCount++
|
||
reconnectCount := room.reconnectCount
|
||
room.mu.Unlock()
|
||
|
||
h.logger.Info("Starting message forwarding from Redis to room",
|
||
zap.String("room_id", roomID),
|
||
zap.String("server_id", h.serverID),
|
||
zap.Int("reconnect_count", reconnectCount),
|
||
)
|
||
} else {
|
||
h.logger.Info("Starting message forwarding from Redis to room",
|
||
zap.String("room_id", roomID),
|
||
zap.String("server_id", h.serverID),
|
||
)
|
||
}
|
||
|
||
for {
|
||
select {
|
||
case <-ctx.Done():
|
||
h.logger.Info("Stopping Redis message forwarding", zap.String("room_id", roomID))
|
||
return
|
||
|
||
case data, ok := <-msgChan:
|
||
if !ok {
|
||
h.logger.Warn("Redis message channel closed for room", zap.String("room_id", roomID))
|
||
return
|
||
}
|
||
|
||
// 1. 日志记录：记录消息类型和长度
|
||
msgType := byte(0)
|
||
if len(data) > 0 {
|
||
msgType = data[0]
|
||
}
|
||
h.logger.Debug("Received message from Redis",
|
||
zap.String("room_id", roomID),
|
||
zap.Int("bytes", len(data)),
|
||
zap.Uint8("msg_type", msgType),
|
||
)
|
||
|
||
// 2. 获取本地房间对象
|
||
h.mu.RLock()
|
||
room, exists := h.rooms[roomID]
|
||
h.mu.RUnlock()
|
||
|
||
if !exists {
|
||
// 这种情况常见于：Redis 消息飞过来时，本地最后一个用户刚好断开删除了房间
|
||
h.logger.Warn("Received Redis message but room does not exist locally",
|
||
zap.String("room_id", roomID),
|
||
)
|
||
continue
|
||
}
|
||
|
||
// 3. 广播给本地所有客户端
|
||
// sender=nil 确保每个人（包括原本发送这条消息的人，如果他在本台机器上）都能收到
|
||
h.broadcastToLocalClients(room, data, nil)
|
||
|
||
// 4. 如果是感知数据 (Awareness)，更新房间缓存
|
||
// 这样后来加入的用户能立即看到其他人的光标
|
||
// if len(data) > 0 && data[0] == 1 {
|
||
// room.mu.Lock()
|
||
// room.lastAwareness = make([]byte, len(data))
|
||
// copy(room.lastAwareness, data)
|
||
// room.mu.Unlock()
|
||
// }
|
||
}
|
||
}
|
||
}
|
||
|
||
// 辅助方法：处理发送失败逻辑，保持代码整洁 (DRY 原则)
|
||
func (c *Client) handleSendFailure() {
|
||
c.failureMu.Lock()
|
||
c.failureCount++
|
||
currentFailures := c.failureCount
|
||
c.failureMu.Unlock()
|
||
|
||
// 直接通过 c.hub 访问日志，代码非常自然
|
||
c.hub.logger.Warn("Failed to send message to client",
|
||
zap.String("clientID", c.ID),
|
||
zap.String("roomID", c.roomID),
|
||
zap.Int("failures", currentFailures),
|
||
zap.Int("max", maxSendFailures),
|
||
)
|
||
|
||
if currentFailures >= maxSendFailures {
|
||
c.hub.logger.Error("Client exceeded max failures, disconnecting",
|
||
zap.String("clientID", c.ID))
|
||
|
||
// 这里的异步处理很正确，防止阻塞当前的广播循环
|
||
go func() {
|
||
c.unregister()
|
||
c.Conn.Close()
|
||
}()
|
||
}
|
||
}
|
||
func (h *Hub) SetFallbackMode(fallback bool) {
|
||
h.mu.Lock()
|
||
// 1. 检查状态是否有变化（幂等性）
|
||
if h.fallbackMode == fallback {
|
||
h.mu.Unlock()
|
||
return
|
||
}
|
||
|
||
// 2. 更新状态并记录日志
|
||
h.fallbackMode = fallback
|
||
if fallback {
|
||
h.logger.Warn("Hub entering FALLBACK MODE (local-only communication)")
|
||
} else {
|
||
h.logger.Info("Hub restored to DISTRIBUTED MODE (cross-server sync enabled)")
|
||
}
|
||
|
||
// 3. 如果是从备用模式恢复 (fallback=false)，需要重新激活所有房间的 Redis 订阅
|
||
if !fallback && h.messagebus != nil {
|
||
h.logger.Info("Recovering Redis subscriptions for existing rooms", zap.Int("room_count", len(h.rooms)))
|
||
|
||
for roomID, room := range h.rooms {
|
||
// 先清理旧的订阅句柄（如果有残留）
|
||
if room.cancel != nil {
|
||
room.cancel()
|
||
room.cancel = nil
|
||
}
|
||
|
||
// 为每个房间重新开启订阅
|
||
ctx, cancel := context.WithCancel(context.Background())
|
||
room.cancel = cancel
|
||
|
||
// 尝试重新订阅
|
||
msgChan, err := h.messagebus.Subscribe(ctx, roomID)
|
||
if err != nil {
|
||
h.logger.Error("Failed to re-subscribe during recovery",
|
||
zap.String("room_id", roomID),
|
||
zap.Error(err),
|
||
)
|
||
cancel()
|
||
room.cancel = nil
|
||
continue
|
||
}
|
||
|
||
// 重新启动转发协程
|
||
go h.startRoomMessageForwarding(ctx, roomID, msgChan)
|
||
|
||
h.logger.Debug("Successfully re-synced room with Redis", zap.String("room_id", roomID))
|
||
}
|
||
}
|
||
h.mu.Unlock()
|
||
}
|
||
func (c *Client) ReadPump() {
|
||
c.Conn.SetReadDeadline(time.Now().Add(pongWait))
|
||
c.Conn.SetPongHandler(func(string) error {
|
||
c.Conn.SetReadDeadline(time.Now().Add(pongWait))
|
||
return nil
|
||
})
|
||
defer func() {
|
||
c.unregister()
|
||
c.Conn.Close()
|
||
}()
|
||
|
||
for {
|
||
messageType, message, err := c.Conn.ReadMessage()
|
||
if err != nil {
|
||
if websocket.IsUnexpectedCloseError(err, websocket.CloseGoingAway, websocket.CloseAbnormalClosure) {
|
||
c.hub.logger.Warn("Unexpected WebSocket close",
|
||
zap.String("client_id", c.ID),
|
||
zap.Error(err))
|
||
}
|
||
break
|
||
}
|
||
|
||
// 1. Sniff Yjs client IDs from awareness messages (before broadcast)
|
||
if messageType == websocket.BinaryMessage && len(message) > 0 && message[0] == 1 {
|
||
clockMap := SniffYjsClientIDs(message)
|
||
if len(clockMap) > 0 {
|
||
c.idsMu.Lock()
|
||
for id, clock := range clockMap {
|
||
if clock > c.observedYjsIDs[id] {
|
||
c.hub.logger.Debug("Sniffed Yjs client ID",
|
||
zap.String("client_id", c.ID),
|
||
zap.Uint64("yjs_id", id),
|
||
zap.Uint64("clock", clock))
|
||
c.observedYjsIDs[id] = clock
|
||
}
|
||
}
|
||
c.idsMu.Unlock()
|
||
|
||
// Cache awareness in Redis for cross-server sync
|
||
// Use a bounded worker pool to avoid blocking ReadPump on Redis I/O.
|
||
if !c.hub.fallbackMode && c.hub.messagebus != nil {
|
||
clientIDs := make([]uint64, 0, len(clockMap))
|
||
for clientID := range clockMap {
|
||
clientIDs = append(clientIDs, clientID)
|
||
}
|
||
select {
|
||
case c.hub.awarenessQueue <- awarenessItem{
|
||
roomID: c.roomID,
|
||
clientIDs: clientIDs,
|
||
data: message,
|
||
}:
|
||
default:
|
||
c.hub.logger.Warn("Awareness queue full, dropping update",
|
||
zap.String("room_id", c.roomID),
|
||
zap.Int("clients", len(clientIDs)))
|
||
}
|
||
}
|
||
}
|
||
}
|
||
|
||
// 2. Permission check - block write operations from view-only users
|
||
if c.Permission == "view" {
|
||
isSyncMessage := len(message) > 0 && message[0] == 0
|
||
isUpdateOp := len(message) > 1 && message[1] == 2
|
||
|
||
if isSyncMessage && isUpdateOp {
|
||
c.hub.logger.Warn("Blocked write from view-only user",
|
||
zap.String("client_id", c.ID))
|
||
continue
|
||
}
|
||
// Allow: Awareness (type=1), SyncStep1/2 (type=0, subtype=0/1)
|
||
}
|
||
|
||
// 3. Broadcast to room
|
||
if messageType == websocket.BinaryMessage {
|
||
select {
|
||
case c.hub.Broadcast <- &Message{
|
||
RoomID: c.roomID,
|
||
Data: message,
|
||
sender: c,
|
||
}:
|
||
default:
|
||
c.hub.logger.Warn("Hub broadcast channel full, dropping message",
|
||
zap.String("client_id", c.ID))
|
||
}
|
||
}
|
||
}
|
||
}
|
||
|
||
func (c *Client) WritePump() {
|
||
ticker := time.NewTicker(pingPeriod)
|
||
defer func() {
|
||
ticker.Stop()
|
||
c.unregister()
|
||
c.Conn.Close()
|
||
}()
|
||
|
||
for {
|
||
select {
|
||
case message, ok := <-c.send:
|
||
c.Conn.SetWriteDeadline(time.Now().Add(writeWait))
|
||
if !ok {
|
||
c.Conn.WriteMessage(websocket.CloseMessage, []byte{})
|
||
return
|
||
}
|
||
|
||
if err := c.Conn.WriteMessage(websocket.BinaryMessage, message); err != nil {
|
||
c.hub.logger.Warn("Error writing message to client",
|
||
zap.String("client_id", c.ID),
|
||
zap.Error(err))
|
||
return
|
||
}
|
||
|
||
// P2 fix: write coalescing — drain all queued messages in a tight loop
|
||
for {
|
||
select {
|
||
case extra, ok := <-c.send:
|
||
if !ok {
|
||
c.Conn.WriteMessage(websocket.CloseMessage, []byte{})
|
||
return
|
||
}
|
||
c.Conn.SetWriteDeadline(time.Now().Add(writeWait))
|
||
if err := c.Conn.WriteMessage(websocket.BinaryMessage, extra); err != nil {
|
||
return
|
||
}
|
||
default:
|
||
break
|
||
}
|
||
if len(c.send) == 0 {
|
||
break
|
||
}
|
||
}
|
||
|
||
case <-ticker.C:
|
||
c.Conn.SetWriteDeadline(time.Now().Add(writeWait))
|
||
if err := c.Conn.WriteMessage(websocket.PingMessage, nil); err != nil {
|
||
c.hub.logger.Debug("Ping failed for client",
|
||
zap.String("client_id", c.ID),
|
||
zap.Error(err))
|
||
return
|
||
}
|
||
}
|
||
}
|
||
}
|
||
|
||
func NewClient(id string, userID *uuid.UUID, userName string, userAvatar *string, permission string, conn *websocket.Conn, hub *Hub, roomID string) *Client {
|
||
return &Client{
|
||
ID: id,
|
||
UserID: userID,
|
||
UserName: userName,
|
||
UserAvatar: userAvatar,
|
||
Permission: permission,
|
||
Conn: conn,
|
||
send: make(chan []byte, 8192),
|
||
hub: hub,
|
||
roomID: roomID,
|
||
observedYjsIDs: make(map[uint64]uint64),
|
||
}
|
||
}
|
||
|
||
// Enqueue sends a message to the client send buffer (non-blocking).
|
||
// Returns false if the buffer is full.
|
||
func (c *Client) Enqueue(message []byte) bool {
|
||
select {
|
||
case c.send <- message:
|
||
return true
|
||
default:
|
||
if c.hub != nil && c.hub.logger != nil {
|
||
c.hub.logger.Warn("Client send buffer full during replay",
|
||
zap.String("client_id", c.ID),
|
||
zap.String("room_id", c.roomID))
|
||
}
|
||
return false
|
||
}
|
||
}
|
||
func (c *Client) unregister() {
|
||
c.unregisterOnce.Do(func() {
|
||
c.hub.Unregister <- c
|
||
})
|
||
}
|