Skip to content

Commit ecfbc0c

Browse files
chenxicccclaude
chenxiccc
and
claude
committed
fix: 补全广播消息 UpdatedTimestamp、断点续传、广播错误处理
P2-10: 补全 Note/File/Folder 的 delete/rename 广播消息中缺失的 UpdatedTimestamp 字段，使接收端能正确推进 lastXxxSyncTime， 避免多端同步时重复下发已处理的变更。同时为 FolderSyncDeleteMessage 和 FolderSyncRenameMessage DTO 补充该字段，并在 FileSyncUploadMessage 中补充 PathHash 字段供客户端断点续传使用。 P2-6: 引入 atomic.Int32 failCount 追踪连续广播失败次数，第 4 次失败 时主动调用 WriteClose 清理半断开连接，避免 TCP keepalive 超时前的静 默丢包。使用 == 4 而非 > 3 保证并发场景下只触发一次关闭。 Co-Authored-By: Claude Opus 4.7 <noreply@anthropic.com>
1 parent 0404ee7 commit ecfbc0c

6 files changed

Lines changed: 85 additions & 54 deletions

File tree

‎internal/dto/file_dto_ws.go‎

Lines changed: 1 addition & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -26,6 +26,7 @@ type FileSyncEndMessage struct {
2626
// FileSyncUploadMessage 定义服务端通知客户端需要上传文件的消息结构
2727
type FileSyncUploadMessage struct {
2828
Path string `json:"path" example:"Image.png"` // File path // 文件路径
29+
PathHash string `json:"pathHash" example:"fhash123"` // Path hash // 路径哈希值
2930
SessionID string `json:"sessionId" example:"sess_123456"` // Session ID // 会话 ID
3031
ChunkSize int64 `json:"chunkSize" example:"1048576"` // Chunk size // 分块大小
3132
}

‎internal/dto/folder_dto_ws.go‎

Lines changed: 12 additions & 10 deletions
Original file line numberDiff line numberDiff line change
@@ -11,21 +11,23 @@ type FolderSyncEndMessage struct {
1111
// FolderSyncRenameMessage message structure for folder rename during sync
1212
// FolderSyncRenameMessage 同步过程中文件夹重命名的消息结构
1313
type FolderSyncRenameMessage struct {
14-
Path string `json:"path" form:"path" binding:"required" example:"NewFolder"` // New path // 新路径
15-
PathHash string `json:"pathHash" form:"pathHash" example:"nfhash123"` // New path hash // 新路径哈希
16-
Ctime int64 `json:"ctime" form:"ctime" example:"1700000000"` // Creation timestamp // 创建时间戳
17-
Mtime int64 `json:"mtime" form:"mtime" example:"1700000000"` // Modification timestamp // 修改时间戳
18-
OldPath string `json:"oldPath" form:"oldPath" example:"OldFolder"` // Old path // 旧路径
19-
OldPathHash string `json:"oldPathHash" form:"oldPathHash" example:"ofhash456"` // Old path hash // 旧路径哈希
14+
Path string `json:"path" form:"path" binding:"required" example:"NewFolder"` // New path // 新路径
15+
PathHash string `json:"pathHash" form:"pathHash" example:"nfhash123"` // New path hash // 新路径哈希
16+
Ctime int64 `json:"ctime" form:"ctime" example:"1700000000"` // Creation timestamp // 创建时间戳
17+
Mtime int64 `json:"mtime" form:"mtime" example:"1700000000"` // Modification timestamp // 修改时间戳
18+
OldPath string `json:"oldPath" form:"oldPath" example:"OldFolder"` // Old path // 旧路径
19+
OldPathHash string `json:"oldPathHash" form:"oldPathHash" example:"ofhash456"` // Old path hash // 旧路径哈希
20+
UpdatedTimestamp int64 `json:"lastTime" form:"updatedTimestamp" example:"1700000000"` // Record update timestamp // 记录更新时间戳
2021
}
2122

2223
// FolderSyncDeleteMessage message structure for folder deletion during sync
2324
// FolderSyncDeleteMessage 同步期间文件夹删除的消息结构
2425
type FolderSyncDeleteMessage struct {
25-
Path string `json:"path" form:"path" example:"DeletedFolder"` // Folder path // 文件夹路径
26-
PathHash string `json:"pathHash" form:"pathHash" example:"dfhash789"` // Path hash // 路径哈希值
27-
Ctime int64 `json:"ctime" form:"ctime" example:"1700000000"` // Creation timestamp // 创建时间戳
28-
Mtime int64 `json:"mtime" form:"mtime" example:"1700000000"` // Modification timestamp // 修改时间戳
26+
Path string `json:"path" form:"path" example:"DeletedFolder"` // Folder path // 文件夹路径
27+
PathHash string `json:"pathHash" form:"pathHash" example:"dfhash789"` // Path hash // 路径哈希值
28+
Ctime int64 `json:"ctime" form:"ctime" example:"1700000000"` // Creation timestamp // 创建时间戳
29+
Mtime int64 `json:"mtime" form:"mtime" example:"1700000000"` // Modification timestamp // 修改时间戳
30+
UpdatedTimestamp int64 `json:"lastTime" form:"updatedTimestamp" example:"1700000000"` // Record update timestamp // 记录更新时间戳
2931
}
3032

3133
// FolderSyncModifyMessage message content for folder modification or creation during sync

‎internal/routers/websocket_router/ws_file.go‎

Lines changed: 15 additions & 10 deletions
Original file line numberDiff line numberDiff line change
@@ -172,6 +172,7 @@ func (h *FileWSHandler) FileUploadCheck(c *pkgapp.WebsocketClient, msg *pkgapp.W
172172
c.ToResponse(code.Success.WithData(
173173
dto.FileSyncUploadMessage{
174174
Path: session.Path,
175+
PathHash: session.PathHash,
175176
SessionID: session.ID,
176177
ChunkSize: session.ChunkSize,
177178
},
@@ -647,11 +648,12 @@ func (h *FileWSHandler) FileSync(c *pkgapp.WebsocketClient, msg *pkgapp.WebSocke
647648
// 将删除消息广播给其他客户端
648649
c.BroadcastResponse(code.Success.WithData(
649650
dto.FileSyncDeleteMessage{
650-
Path: fileSvc.Path,
651-
PathHash: fileSvc.PathHash,
652-
Ctime: fileSvc.Ctime,
653-
Mtime: fileSvc.Mtime,
654-
Size: fileSvc.Size,
651+
Path: fileSvc.Path,
652+
PathHash: fileSvc.PathHash,
653+
Ctime: fileSvc.Ctime,
654+
Mtime: fileSvc.Mtime,
655+
Size: fileSvc.Size,
656+
UpdatedTimestamp: fileSvc.UpdatedTimestamp,
655657
},
656658
).WithVault(params.Vault), true, dto.FileSyncDelete)
657659

@@ -671,11 +673,12 @@ func (h *FileWSHandler) FileSync(c *pkgapp.WebsocketClient, msg *pkgapp.WebSocke
671673
// 使用现有信息(Path/PathHash)广播删除
672674
c.BroadcastResponse(code.Success.WithData(
673675
dto.FileSyncDeleteMessage{
674-
Path: delFile.Path,
675-
PathHash: delFile.PathHash,
676-
Ctime: 0,
677-
Mtime: 0,
678-
Size: 0,
676+
Path: delFile.Path,
677+
PathHash: delFile.PathHash,
678+
Ctime: 0,
679+
Mtime: 0,
680+
Size: 0,
681+
UpdatedTimestamp: 0,
679682
},
680683
).WithVault(params.Vault), true, dto.FileSyncDelete)
681684
}
@@ -794,6 +797,7 @@ func (h *FileWSHandler) FileSync(c *pkgapp.WebsocketClient, msg *pkgapp.WebSocke
794797
Action: dto.FileUpload,
795798
Data: dto.FileSyncUploadMessage{
796799
Path: session.Path,
800+
PathHash: session.PathHash,
797801
SessionID: session.ID,
798802
ChunkSize: session.ChunkSize,
799803
},
@@ -859,6 +863,7 @@ func (h *FileWSHandler) FileSync(c *pkgapp.WebsocketClient, msg *pkgapp.WebSocke
859863
Action: dto.FileUpload,
860864
Data: dto.FileSyncUploadMessage{
861865
Path: session.Path,
866+
PathHash: session.PathHash,
862867
SessionID: session.ID,
863868
ChunkSize: session.ChunkSize,
864869
},

‎internal/routers/websocket_router/ws_folder.go‎

Lines changed: 27 additions & 22 deletions
Original file line numberDiff line numberDiff line change
@@ -80,10 +80,11 @@ func (h *FolderWSHandler) FolderSync(c *pkgapp.WebsocketClient, msg *pkgapp.WebS
8080
// Broadcast deletion to other clients
8181
c.BroadcastResponse(code.Success.WithData(
8282
dto.FolderSyncDeleteMessage{
83-
Path: folder.Path,
84-
PathHash: folder.PathHash,
85-
Ctime: folder.Ctime,
86-
Mtime: folder.Mtime,
83+
Path: folder.Path,
84+
PathHash: folder.PathHash,
85+
Ctime: folder.Ctime,
86+
Mtime: folder.Mtime,
87+
UpdatedTimestamp: folder.UpdatedTimestamp,
8788
},
8889
).WithVault(params.Vault).WithContext(params.Context), true, dto.FolderSyncDelete)
8990
} else {
@@ -95,10 +96,11 @@ func (h *FolderWSHandler) FolderSync(c *pkgapp.WebsocketClient, msg *pkgapp.WebS
9596
// Broadcast deletion with available info
9697
c.BroadcastResponse(code.Success.WithData(
9798
dto.FolderSyncDeleteMessage{
98-
Path: delFolder.Path,
99-
PathHash: delFolder.PathHash,
100-
Ctime: 0,
101-
Mtime: 0,
99+
Path: delFolder.Path,
100+
PathHash: delFolder.PathHash,
101+
Ctime: 0,
102+
Mtime: 0,
103+
UpdatedTimestamp: 0,
102104
},
103105
).WithVault(params.Vault).WithContext(params.Context), true, dto.FolderSyncDelete)
104106
}
@@ -161,10 +163,11 @@ func (h *FolderWSHandler) FolderSync(c *pkgapp.WebsocketClient, msg *pkgapp.WebS
161163
messageQueue = append(messageQueue, dto.WSQueuedMessage{
162164
Action: dto.FolderSyncDelete,
163165
Data: dto.FolderSyncDeleteMessage{
164-
Path: folder.Path,
165-
PathHash: folder.PathHash,
166-
Ctime: folder.Ctime,
167-
Mtime: folder.Mtime,
166+
Path: folder.Path,
167+
PathHash: folder.PathHash,
168+
Ctime: folder.Ctime,
169+
Mtime: folder.Mtime,
170+
UpdatedTimestamp: folder.UpdatedTimestamp,
168171
},
169172
})
170173
needDeleteCount++
@@ -286,10 +289,11 @@ func (h *FolderWSHandler) FolderDelete(c *pkgapp.WebsocketClient, msg *pkgapp.We
286289
}), string(dto.FolderDeleteAck))
287290
c.BroadcastResponse(code.Success.WithData(
288291
dto.FolderSyncDeleteMessage{
289-
Path: folder.Path,
290-
PathHash: folder.PathHash,
291-
Ctime: folder.Ctime,
292-
Mtime: folder.Mtime,
292+
Path: folder.Path,
293+
PathHash: folder.PathHash,
294+
Ctime: folder.Ctime,
295+
Mtime: folder.Mtime,
296+
UpdatedTimestamp: folder.UpdatedTimestamp,
293297
},
294298
).WithVault(params.Vault), true, dto.FolderSyncDelete)
295299
}
@@ -333,12 +337,13 @@ func (h *FolderWSHandler) FolderRename(c *pkgapp.WebsocketClient, msg *pkgapp.We
333337
}
334338

335339
c.BroadcastResponse(code.Success.WithData(dto.FolderSyncRenameMessage{
336-
Path: newFolder.Path,
337-
PathHash: newFolder.PathHash,
338-
Ctime: newFolder.Ctime,
339-
Mtime: newFolder.Mtime,
340-
OldPath: oldFolder.Path,
341-
OldPathHash: oldFolder.PathHash,
340+
Path: newFolder.Path,
341+
PathHash: newFolder.PathHash,
342+
Ctime: newFolder.Ctime,
343+
Mtime: newFolder.Mtime,
344+
OldPath: oldFolder.Path,
345+
OldPathHash: oldFolder.PathHash,
346+
UpdatedTimestamp: newFolder.UpdatedTimestamp,
342347
}).WithVault(params.Vault), true, dto.FolderSyncRename)
343348

344349
}

‎internal/routers/websocket_router/ws_note.go‎

Lines changed: 12 additions & 10 deletions
Original file line numberDiff line numberDiff line change
@@ -748,11 +748,12 @@ func (h *NoteWSHandler) NoteSync(c *pkgapp.WebsocketClient, msg *pkgapp.WebSocke
748748
// 将删除消息广播给其他客户端
749749
c.BroadcastResponse(code.Success.WithData(
750750
dto.NoteSyncDeleteMessage{
751-
Path: note.Path,
752-
PathHash: note.PathHash,
753-
Ctime: note.Ctime,
754-
Mtime: note.Mtime,
755-
Size: note.Size,
751+
Path: note.Path,
752+
PathHash: note.PathHash,
753+
Ctime: note.Ctime,
754+
Mtime: note.Mtime,
755+
Size: note.Size,
756+
UpdatedTimestamp: note.UpdatedTimestamp,
756757
},
757758
).WithVault(params.Vault), true, dto.NoteSyncDelete)
758759

@@ -772,11 +773,12 @@ func (h *NoteWSHandler) NoteSync(c *pkgapp.WebsocketClient, msg *pkgapp.WebSocke
772773
// 使用现有信息(Path/PathHash)广播删除
773774
c.BroadcastResponse(code.Success.WithData(
774775
dto.NoteSyncDeleteMessage{
775-
Path: delNote.Path,
776-
PathHash: delNote.PathHash,
777-
Ctime: 0,
778-
Mtime: 0,
779-
Size: 0,
776+
Path: delNote.Path,
777+
PathHash: delNote.PathHash,
778+
Ctime: 0,
779+
Mtime: 0,
780+
Size: 0,
781+
UpdatedTimestamp: 0,
780782
},
781783
).WithVault(params.Vault), true, dto.NoteSyncDelete)
782784
}

‎pkg/app/websocket.go‎

Lines changed: 18 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -7,6 +7,7 @@ import (
77
"strconv"
88
"strings"
99
"sync"
10+
"sync/atomic"
1011
"time"
1112

1213
"github.com/google/uuid"
@@ -263,6 +264,7 @@ type WebsocketClient struct {
263264
DiffMergePaths map[string]DiffMergeEntry // File paths needing merging // 需要合并的文件路径，包含创建时间用于超时清理
264265
DiffMergePathsMu sync.RWMutex // Mutex lock to prevent concurrency conflicts // 互斥锁，防止并发冲突
265266
OfflineSyncStrategy string // Offline device sync strategy // 离线设备同步策略 "newTimeMerge" | "ignoreTimeMerge"
267+
failCount atomic.Int32 // Consecutive broadcast failure counter; connection closed when exceeding threshold // 连续广播失败计数器，超过阈值时主动关闭连接
266268
}
267269

268270
// initContext initializes the context for the WebSocket connection
@@ -560,7 +562,15 @@ func (c *WebsocketClient) sendBroadcast(payload []byte, isExcludeSelf bool) {
560562
continue
561563
}
562564

563-
_ = b.Broadcast(uc.conn)
565+
// Track consecutive broadcast failures and close half-broken connections proactively.
566+
// 追踪连续广播失败次数，主动关闭半断开的连接（TCP keepalive 未超时但已无法通信）。
567+
if err := b.Broadcast(uc.conn); err != nil {
568+
if uc.failCount.Add(1) == 4 {
569+
uc.conn.WriteClose(1000, []byte("broadcast failed"))
570+
}
571+
} else {
572+
uc.failCount.Store(0)
573+
}
564574
}
565575
}
566576

@@ -1171,6 +1181,12 @@ func (w *WebsocketServer) BroadcastToUser(uid int64, code *code.Code, action str
11711181
if uc.conn == nil {
11721182
continue
11731183
}
1174-
_ = b.Broadcast(uc.conn)
1184+
if err := b.Broadcast(uc.conn); err != nil {
1185+
if uc.failCount.Add(1) == 4 {
1186+
uc.conn.WriteClose(1000, []byte("broadcast failed"))
1187+
}
1188+
} else {
1189+
uc.failCount.Store(0)
1190+
}
11751191
}
11761192
}

0 commit comments

Comments
 (0)