fix(sync): 同步服务修复——版本冲突判定/WS查询洪泛/Resolve安全/CRDT幂等/广播完整性

- SYNC-H1: Push 冲突判定收紧为 client version <= server version——相同版本号不同内容的静默覆盖改为返回冲突（双设备并发编辑不再丢数据；幂等重试由 op.ID 去重覆盖）
- SYNC-H2: WS sync_request 每连接在飞查询上限 2（信号槽），单连接洪泛不再打满数据库连接池（50）拖垮全部用户
- SYNC-M1: Resolve 增加 FOR UPDATE 行锁 + 版本必须大于服务端版本（拒绝版本回退）；成功后广播到其他在线设备（实时收敛）
- SYNC-M2: Push 广播条件从 opSeqNoByID 非空改为 accepted 非空——无 ID 操作（fallback ID）落库后同样广播
- SYNC-M3: CRDTPush changeset SHA-256 哈希幂等去重（(user_id, changeset_hash) 唯一索引 + 并发兜底），网络重试不再重复落库膨胀
- SYNC-L2: Pull/CRDTPull 的 deviceId 白名单校验（防回环与超长参数）

验证: go build + go vet + go test 全部通过

Author: Aparencia

server/sync-service/handlers/crdt.go

@@ -5,6 +5,8 @@
 package handlers
 
 import (
+	"crypto/sha256"
+	"encoding/hex"
 	"net/http"
 	"strconv"
 	"strings"
@@ -54,21 +56,45 @@ func CRDTPush(c *gin.Context) {
 	// M7: 整个循环包裹在单个事务内
 	txErr := models.DB.Transaction(func(tx *gorm.DB) error {
 		for _, ch := range req.Changes {
+			// SYNC-M3: changeset 哈希幂等去重——网络重试的重复 changeset
+			// 不再重复分配序号与落库（(user_id, changeset_hash) 唯一索引兜底）
+			if ch.Changeset != "" {
+				hash := sha256.Sum256([]byte(ch.Changeset))
+				hashHex := hex.EncodeToString(hash[:])
+				var dupCount int64
+				if err := tx.Model(&models.CRDTChange{}).
+					Where("user_id = ? AND changeset_hash = ?", userID, hashHex).
+					Count(&dupCount).Error; err != nil {
+					return err
+				}
+				if dupCount > 0 {
+					// 已处理过：视为已接受（幂等），不重复落库
+					accepted = append(accepted, ch.Seq)
+					continue
+				}
+			}
+
 			seqNo, err := nextSeqNo(tx)
 			if err != nil {
 				return err
 			}
 
 			if err := tx.Create(&models.CRDTChange{
-				ServerSeqNo: seqNo,
-				DeviceID:    req.DeviceID,
-				UserID:      userID,
-				TableName:   ch.TableName,
-				EntityID:    ch.EntityID,
-				Changeset:   ch.Changeset,
-				Operation:   ch.Operation,
-				CreatedAt:   ch.CreatedAt,
+				ServerSeqNo:   seqNo,
+				DeviceID:      req.DeviceID,
+				UserID:        userID,
+				TableName:     ch.TableName,
+				EntityID:      ch.EntityID,
+				Changeset:     ch.Changeset,
+				ChangesetHash: hashChangeSet(ch.Changeset),
+				Operation:     ch.Operation,
+				CreatedAt:     ch.CreatedAt,
 			}).Error; err != nil {
+				// 唯一索引兜底：并发重复推送视为已处理
+				if isUniqueViolation(err) && ch.Changeset != "" {
+					accepted = append(accepted, ch.Seq)
+					continue
+				}
 				return err
 			}
 			accepted = append(accepted, ch.Seq)
@@ -88,6 +114,15 @@ func CRDTPush(c *gin.Context) {
 	})
 }
 
+// SYNC-M3: 计算 changeset 的 SHA-256 hex（空串返回空串，不参与幂等）
+func hashChangeSet(changeset string) string {
+	if changeset == "" {
+		return ""
+	}
+	hash := sha256.Sum256([]byte(changeset))
+	return hex.EncodeToString(hash[:])
+}
+
 // ─── CRDT Changeset Pull (GET /api/v1/sync/crdt/changes) ──────────────────────
 
 // CRDTPull returns all CRDT changesets since `since` (exclusive) that were
@@ -96,6 +131,11 @@ func CRDTPush(c *gin.Context) {
 func CRDTPull(c *gin.Context) {
 	userID := c.GetString("user_id")
 	deviceID := c.Query("deviceId")
+	// SYNC-L2: deviceId 白名单校验——空值拉回本设备变更造成回环
+	if !isValidDeviceID(deviceID) {
+		c.JSON(http.StatusBadRequest, gin.H{"error": "invalid deviceId: must match [A-Za-z0-9_-]{1,64}"})
+		return
+	}
 	sinceStr := c.DefaultQuery("since", "0")
 	since, err := strconv.ParseInt(sinceStr, 10, 64)
 	if err != nil {

server/sync-service/handlers/sync.go

@@ -91,8 +91,11 @@ func Push(c *gin.Context) {
 			}
 			entityExists := result.RowsAffected > 0
 
-			// M1: 版本比较在事务锁内执行
-			if entityExists && op.Version < ev.Version {
+			// SYNC-H1: 冲突判定收紧为 client version <= server version。
+			// 原实现仅拒绝 <，两台设备基于同一版本各自编辑会产生相同版本号
+			// 不同内容，后到者静默覆盖先到者（数据丢失且无冲突提示）。
+			// 相同版本的幂等重试由 op.ID 去重处理，不受影响。
+			if entityExists && op.Version <= ev.Version {
 				// Conflict: client is behind server.
 				conflicts = append(conflicts, ConflictInfo{
 					EntityType:    op.EntityType,
@@ -190,11 +193,26 @@ func Push(c *gin.Context) {
 
 	// Broadcast accepted operations to the user's other online devices via WebSocket.
 	// M11: 广播同时携带 ServerSeqNo（Pull 游标）与 Version（实体版本），语义各自独立。
-	if len(accepted) > 0 && len(opSeqNoByID) > 0 {
+	// SYNC-M2: 广播条件从 opSeqNoByID 非空改为 accepted 非空——
+	// 全部操作无 ID（服务端 fallback ID）时变更已落库但不再被广播；
+	// 空 ID 操作同样需要广播（其 ServerSeqNo 为 0，客户端按实体版本收敛）。
+	if len(accepted) > 0 {
+		skippedSet := make(map[string]bool, len(skipped))
+		for _, id := range skipped {
+			skippedSet[id] = true
+		}
+		conflictKeys := make(map[string]bool, len(conflicts))
+		for _, cf := range conflicts {
+			conflictKeys[cf.EntityType+":"+cf.EntityID] = true
+		}
+
 		wsOps := make([]WSOperationPayload, 0, len(accepted))
 		for _, op := range req.Operations {
-			seqNo, ok := opSeqNoByID[op.ID]
-			if !ok {
+			// 跳过被去重跳过或冲突的操作
+			if op.ID != "" && skippedSet[op.ID] {
+				continue
+			}
+			if conflictKeys[op.EntityType+":"+op.EntityID] {
 				continue
 			}
 			wsOps = append(wsOps, WSOperationPayload{
@@ -203,7 +221,7 @@ func Push(c *gin.Context) {
 				Operation:   op.Operation,
 				Data:        op.Payload,
 				Version:     op.Version,
-				ServerSeqNo: seqNo,
+				ServerSeqNo: opSeqNoByID[op.ID], // 空 ID 操作未登记序号时为 0
 				DeviceID:    req.DeviceID,
 			})
 		}

server/sync-service/handlers/sync_query.go

@@ -20,6 +20,12 @@ import (
 func Pull(c *gin.Context) {
 	userID := c.GetString("user_id")
 	deviceID := c.Query("deviceId")
+	// SYNC-L2: deviceId 白名单校验（与 WS 入口一致）——空值会让
+	// device_id != '' 拉回本设备变更造成回环；超长参数增加索引扫描成本
+	if !isValidDeviceID(deviceID) {
+		c.JSON(http.StatusBadRequest, gin.H{"error": "invalid deviceId: must match [A-Za-z0-9_-]{1,64}"})
+		return
+	}
 	sinceVersionStr := c.DefaultQuery("sinceVersion", "0")
 	sinceVersion, err := strconv.ParseInt(sinceVersionStr, 10, 64)
 	if err != nil {

server/sync-service/handlers/sync_resolve.go

@@ -5,6 +5,8 @@
 package handlers
 
 import (
+	"errors"
+	"log"
 	"net/http"
 	"strings"
 	"time"
@@ -13,6 +15,7 @@ import (
 
 	"github.com/gin-gonic/gin"
 	"gorm.io/gorm"
+	"gorm.io/gorm/clause"
 )
 
 type resolveRequest struct {
@@ -56,14 +59,26 @@ func Resolve(c *gin.Context) {
 		}
 
 		txErr := models.DB.Transaction(func(tx *gorm.DB) error {
+			// SYNC-M1: 版本校验 + FOR UPDATE 行锁——与 Push 的 M1 防护一致，
+			// 拒绝回退版本（客户端可提交任意小版本号制造数据回退）；
+			// 加锁消除 TOCTOU 竞态（另一设备并发推送时基于过期版本覆盖）
+			var ev models.EntityVersion
+			res := tx.Clauses(clause.Locking{Strength: "UPDATE"}).
+				Where("user_id = ? AND entity_type = ? AND entity_id = ?", userID, req.EntityType, req.EntityID).
+				First(&ev)
+			if res.Error != nil && res.Error != gorm.ErrRecordNotFound {
+				return res.Error
+			}
+			if res.RowsAffected > 0 && req.Version <= ev.Version {
+				return errors.New("resolve rejected: client version must be newer than server version")
+			}
+
 			seqNo, err := nextSeqNo(tx)
 			if err != nil {
 				return err
 			}
 
 			// Upsert EntityVersion.
-			var ev models.EntityVersion
-			res := tx.Where("user_id = ? AND entity_type = ? AND entity_id = ?", userID, req.EntityType, req.EntityID).First(&ev)
 			if res.RowsAffected > 0 {
 				if err := tx.Model(&ev).Updates(map[string]interface{}{
 					"version": req.Version,
@@ -98,10 +113,25 @@ func Resolve(c *gin.Context) {
 		})
 
 		if txErr != nil {
+			log.Printf("Resolve failed: %v", txErr)
 			c.JSON(http.StatusInternalServerError, gin.H{"error": txErr.Error()})
 			return
 		}
 
+		// SYNC-M1: 广播解决结果到其他在线设备（原实现不广播，
+		// 其他设备只能等下一次 Pull 才收敛，实时性延迟）
+		BroadcastOperation(userID, req.DeviceID, []WSOperationPayload{
+			{
+				EntityType:  req.EntityType,
+				EntityID:    req.EntityID,
+				Operation:   "update",
+				Data:        req.Data,
+				Version:     req.Version,
+				ServerSeqNo: 0, // Resolve 的广播无明确序号，客户端按实体版本收敛
+				DeviceID:    req.DeviceID,
+			},
+		})
+
 	case "remote":
 		// Server wins: no changes needed; client will pull latest.
 

server/sync-service/handlers/websocket.go

@@ -156,6 +156,8 @@ func HandleWebSocketWithGin(c *gin.Context, userID string, deviceID string) {
 		DeviceID: deviceID,
 		Conn:     conn,
 		Send:     make(chan []byte, 256),
+		// SYNC-H2: 每连接最多 2 个在飞 sync_request 查询
+		syncReqSlots: make(chan struct{}, 2),
 	}
 
 	// M12: 每用户连接数上限检查，超限则拒绝（发送 ClosePolicyViolation 帧后关闭）

server/sync-service/handlers/ws_connection.go

@@ -40,6 +40,10 @@ type WSConnection struct {
 	Send     chan []byte
 	closed   bool
 	mu       sync.Mutex
+
+	// SYNC-H2: sync_request 在飞查询信号槽（容量 2）——客户端可连续发送
+	// 任意数量 sync_request，每个都启动 goroutine 查库会打满数据库连接池
+	syncReqSlots chan struct{}
 }
 
 // close safely marks the connection as closed and closes the underlying socket.
@@ -121,7 +125,17 @@ func (c *WSConnection) readPump() {
 			}
 			// M10: 查库改为异步执行（goroutine），结果经 Send 通道回投，
 			// 避免慢查询阻塞 readPump 导致心跳超时。
-			go c.handleSyncRequest(payload.SinceVersion)
+			// SYNC-H2: 在飞查询超限时丢弃本次请求（客户端轮询会重发），
+			// 防止单连接洪泛打满数据库连接池（上限 50）拖垮全部用户
+			select {
+			case c.syncReqSlots <- struct{}{}:
+				go func() {
+					defer func() { <-c.syncReqSlots }()
+					c.handleSyncRequest(payload.SinceVersion)
+				}()
+			default:
+				log.Printf("[ws] sync_request dropped (in-flight limit reached) user=%s device=%s", c.UserID, c.DeviceID)
+			}
 
 		case "operation":
 			// Client-pushed operation over WebSocket (future enhancement).

server/sync-service/models/models.go

@@ -42,14 +42,17 @@ type GlobalSeqNo struct {
 
 // CRDTChange stores CRDT changesets uploaded by clients.
 // Each row is an immutable, append-only change record keyed by a global sequence number.
+// ChangesetHash 为 changeset 的 SHA-256（hex），配合 (user_id, changeset_hash) 唯一
+// 索引实现 CRDT 推送幂等（SYNC-M3）：网络重试的重复 changeset 不再重复落库。
 type CRDTChange struct {
-	ID          uint   `gorm:"primaryKey"`
-	ServerSeqNo int64  `gorm:"uniqueIndex;not null" json:"serverSeqNo"`
-	DeviceID    string `gorm:"index;not null" json:"deviceId"`
-	UserID      string `gorm:"index:idx_crdt_user;not null" json:"userId"`
-	TableName   string `gorm:"index;not null" json:"tableName"`
-	EntityID    string `gorm:"not null" json:"entityId"`
-	Changeset   string `gorm:"type:text;not null" json:"changeset"` // base64-encoded Automerge changes
-	Operation   string `gorm:"not null" json:"operation"`           // create|update|delete
-	CreatedAt   string `gorm:"not null" json:"createdAt"`
+	ID            uint   `gorm:"primaryKey"`
+	ServerSeqNo   int64  `gorm:"uniqueIndex;not null" json:"serverSeqNo"`
+	DeviceID      string `gorm:"index;not null" json:"deviceId"`
+	UserID        string `gorm:"index:idx_crdt_user;uniqueIndex:idx_crdt_user_hash,priority:1;not null" json:"userId"`
+	TableName     string `gorm:"index;not null" json:"tableName"`
+	EntityID      string `gorm:"not null" json:"entityId"`
+	Changeset     string `gorm:"type:text;not null" json:"changeset"` // base64-encoded Automerge changes
+	ChangesetHash string `gorm:"uniqueIndex:idx_crdt_user_hash,priority:2;not null" json:"-"` // sha256(changeset)
+	Operation     string `gorm:"not null" json:"operation"`             // create|update|delete
+	CreatedAt     string `gorm:"not null" json:"createdAt"`
 }