From 9dab6a5039b5043d5e5127aa08d9614d59cbbcc2 Mon Sep 17 00:00:00 2001 From: naiba Date: Sat, 19 Sep 2026 10:44:17 +0000 Subject: [PATCH] fix: harden notification cache synchronization --- cmd/dashboard/controller/notification.go | 6 + .../notification_body_limit_test.go | 25 ++++ service/singleton/notification.go | 123 +++++++++-------- service/singleton/notification_test.go | 126 ++++++++++++++++++ 4 files changed, 225 insertions(+), 55 deletions(-) create mode 100644 cmd/dashboard/controller/notification_body_limit_test.go create mode 100644 service/singleton/notification_test.go diff --git a/cmd/dashboard/controller/notification.go b/cmd/dashboard/controller/notification.go index 43403e34..0264b339 100644 --- a/cmd/dashboard/controller/notification.go +++ b/cmd/dashboard/controller/notification.go @@ -1,6 +1,7 @@ package controller import ( + "net/http" "slices" "strconv" @@ -12,6 +13,8 @@ import ( "github.com/nezhahq/nezha/service/singleton" ) +const notificationBatchDeleteMaxBodyBytes = 1 << 20 + // List notification // @Summary List notification // @Security BearerAuth @@ -174,6 +177,9 @@ func updateNotification(c *gin.Context) (any, error) { // @Router /batch-delete/notification [post] func batchDeleteNotification(c *gin.Context) (any, error) { var n []uint64 + if c.Request != nil && c.Request.Body != nil { + c.Request.Body = http.MaxBytesReader(c.Writer, c.Request.Body, notificationBatchDeleteMaxBodyBytes) + } if err := c.ShouldBindJSON(&n); err != nil { return nil, err } diff --git a/cmd/dashboard/controller/notification_body_limit_test.go b/cmd/dashboard/controller/notification_body_limit_test.go new file mode 100644 index 00000000..4501cee9 --- /dev/null +++ b/cmd/dashboard/controller/notification_body_limit_test.go @@ -0,0 +1,25 @@ +package controller + +import ( + "bytes" + "net/http" + "net/http/httptest" + "strings" + "testing" + + "github.com/gin-gonic/gin" +) + +func TestBatchDeleteNotificationCapsRequestBody(t *testing.T) { + body := append(bytes.Repeat([]byte{' '}, notificationBatchDeleteMaxBodyBytes+1), '[', ']') + req := httptest.NewRequest(http.MethodPost, "/api/v1/batch-delete/notification", bytes.NewReader(body)) + req.Header.Set("Content-Type", "application/json") + w := httptest.NewRecorder() + c, _ := gin.CreateTestContext(w) + c.Request = req + + _, err := batchDeleteNotification(c) + if err == nil || !strings.Contains(err.Error(), "request body too large") { + t.Fatalf("oversized request body was not rejected by MaxBytesReader: %v", err) + } +} diff --git a/service/singleton/notification.go b/service/singleton/notification.go index 0b358ab1..31bf3fe2 100644 --- a/service/singleton/notification.go +++ b/service/singleton/notification.go @@ -81,20 +81,29 @@ func NewNotificationClass() *NotificationClass { } func (c *NotificationClass) Update(n *model.Notification) { - c.listMu.Lock() + func() { + c.listMu.Lock() + defer c.listMu.Unlock() - _, ok := c.list[n.ID] - c.list[n.ID] = n - - if ok { - if gids, ok := c.idToGroupList[n.ID]; ok { - for gid := range gids { - c.groupToIDList[gid][n.ID] = n - } + _, ok := c.list[n.ID] + c.list[n.ID] = n + if !ok { + return } - } - c.listMu.Unlock() + gids := c.idToGroupList[n.ID] + for gid := range gids { + group, exists := c.groupToIDList[gid] + if !exists { + delete(gids, gid) + continue + } + group[n.ID] = n + } + if len(gids) == 0 { + delete(c.idToGroupList, n.ID) + } + }() c.sortList() } @@ -102,71 +111,69 @@ func (c *NotificationClass) UpdateGroup(ng *model.NotificationGroup, ngn []uint6 c.groupMu.Lock() defer c.groupMu.Unlock() - _, ok := c.groupList[ng.ID] c.groupList[ng.ID] = ng.Name c.listMu.Lock() defer c.listMu.Unlock() - if !ok { - c.groupToIDList[ng.ID] = make(map[uint64]*model.Notification, len(ngn)) - for _, n := range ngn { - if c.idToGroupList[n] == nil { - c.idToGroupList[n] = make(map[uint64]struct{}) - } - c.idToGroupList[n][ng.ID] = struct{}{} - c.groupToIDList[ng.ID][n] = c.list[n] - } - } else { - oldList := make(map[uint64]struct{}) - for nid := range c.groupToIDList[ng.ID] { - oldList[nid] = struct{}{} - } - c.groupToIDList[ng.ID] = make(map[uint64]*model.Notification) - for _, nid := range ngn { - c.groupToIDList[ng.ID][nid] = c.list[nid] - if c.idToGroupList[nid] == nil { - c.idToGroupList[nid] = make(map[uint64]struct{}) - } - c.idToGroupList[nid][ng.ID] = struct{}{} + oldList := c.groupToIDList[ng.ID] + newList := make(map[uint64]*model.Notification, len(ngn)) + for _, nid := range ngn { + n, ok := c.list[nid] + if !ok { + continue } + newList[nid] = n + if c.idToGroupList[nid] == nil { + c.idToGroupList[nid] = make(map[uint64]struct{}) + } + c.idToGroupList[nid][ng.ID] = struct{}{} + } - for oldID := range oldList { - if _, ok := c.groupToIDList[ng.ID][oldID]; !ok { - delete(c.groupToIDList[oldID], ng.ID) - if len(c.idToGroupList[oldID]) == 0 { - delete(c.idToGroupList, oldID) - } - } + for oldID := range oldList { + if _, ok := newList[oldID]; ok { + continue + } + delete(c.idToGroupList[oldID], ng.ID) + if len(c.idToGroupList[oldID]) == 0 { + delete(c.idToGroupList, oldID) } } + c.groupToIDList[ng.ID] = newList } func (c *NotificationClass) Delete(idList []uint64) { - c.listMu.Lock() + func() { + c.listMu.Lock() + defer c.listMu.Unlock() - for _, id := range idList { - delete(c.list, id) - // 如果绑定了通知组才删除 - if gids, ok := c.idToGroupList[id]; ok { - for gid := range gids { - delete(c.groupToIDList[gid], id) + for _, id := range idList { + delete(c.list, id) + // 如果绑定了通知组才删除 + if gids, ok := c.idToGroupList[id]; ok { + for gid := range gids { + delete(c.groupToIDList[gid], id) + } delete(c.idToGroupList, id) } } - } - - c.listMu.Unlock() + }() c.sortList() } func (c *NotificationClass) DeleteGroup(gids []uint64) { - c.listMu.Lock() - defer c.listMu.Unlock() c.groupMu.Lock() defer c.groupMu.Unlock() + c.listMu.Lock() + defer c.listMu.Unlock() for _, gid := range gids { + for nid := range c.groupToIDList[gid] { + delete(c.idToGroupList[nid], gid) + if len(c.idToGroupList[nid]) == 0 { + delete(c.idToGroupList, nid) + } + } delete(c.groupList, gid) delete(c.groupToIDList, gid) } @@ -234,13 +241,19 @@ func (c *NotificationClass) SendNotification(notificationGroupID uint64, desc st return } } - // 向该通知方式组的所有通知方式发出通知 + // Copy the group under the lock. Webhook delivery can take minutes and must + // not block notification or notification-group updates while it is in flight. c.listMu.RLock() - defer c.listMu.RUnlock() + notifications := make([]*model.Notification, 0, len(c.groupToIDList[notificationGroupID])) for _, n := range c.groupToIDList[notificationGroupID] { + notifications = append(notifications, n) + } + c.listMu.RUnlock() + + for _, n := range notifications { log.Printf("NEZHA>> Try to notify %s", n.Name) } - for _, n := range c.groupToIDList[notificationGroupID] { + for _, n := range notifications { ns := model.NotificationServerBundle{ Notification: n, Server: nil, diff --git a/service/singleton/notification_test.go b/service/singleton/notification_test.go new file mode 100644 index 00000000..5e08de31 --- /dev/null +++ b/service/singleton/notification_test.go @@ -0,0 +1,126 @@ +package singleton + +import ( + "testing" + "time" + + "github.com/nezhahq/nezha/model" +) + +func newNotificationClassWithItems(items ...*model.Notification) *NotificationClass { + nc := NewEmptyNotificationClassForTest() + for _, item := range items { + nc.InsertForTest(item) + } + return nc +} + +func TestNotificationClassDeleteGroupCleansReverseIndex(t *testing.T) { + n := &model.Notification{Common: model.Common{ID: 1, UserID: 2}, Name: "hook"} + nc := newNotificationClassWithItems(n) + group := &model.NotificationGroup{Common: model.Common{ID: 10, UserID: 2}, Name: "group"} + nc.UpdateGroup(group, []uint64{n.ID}) + + nc.DeleteGroup([]uint64{group.ID}) + + if _, ok := nc.groupToIDList[group.ID]; ok { + t.Fatalf("forward index still contains deleted group %d", group.ID) + } + if groups, ok := nc.idToGroupList[n.ID]; ok { + if _, stale := groups[group.ID]; stale { + t.Fatalf("reverse index for notification %d still contains deleted group %d", n.ID, group.ID) + } + } +} + +func TestNotificationClassUpdateGroupCleansRemovedMembership(t *testing.T) { + first := &model.Notification{Common: model.Common{ID: 1, UserID: 2}, Name: "first"} + second := &model.Notification{Common: model.Common{ID: 2, UserID: 2}, Name: "second"} + nc := newNotificationClassWithItems(first, second) + group := &model.NotificationGroup{Common: model.Common{ID: 10, UserID: 2}, Name: "group"} + nc.UpdateGroup(group, []uint64{first.ID, second.ID}) + + nc.UpdateGroup(group, []uint64{second.ID}) + + if groups, ok := nc.idToGroupList[first.ID]; ok { + if _, stale := groups[group.ID]; stale { + t.Fatalf("reverse index for notification %d still contains group %d", first.ID, group.ID) + } + } + if _, ok := nc.idToGroupList[second.ID][group.ID]; !ok { + t.Fatalf("reverse index lost retained membership of notification %d in group %d", second.ID, group.ID) + } +} + +func TestNotificationClassUpdateRepairsOrphanedReverseIndex(t *testing.T) { + n := &model.Notification{Common: model.Common{ID: 1, UserID: 2}, Name: "hook"} + nc := newNotificationClassWithItems(n) + nc.idToGroupList[n.ID] = map[uint64]struct{}{99: {}} + + n.Name = "updated" + nc.Update(n) + + if groups, ok := nc.idToGroupList[n.ID]; ok && len(groups) != 0 { + t.Fatalf("orphaned reverse memberships were not removed: %v", groups) + } +} + +func TestNotificationClassGroupMutationsDoNotDeadlock(t *testing.T) { + nc := NewEmptyNotificationClassForTest() + nc.listMu.RLock() + readLockHeld := true + defer func() { + if readLockHeld { + nc.listMu.RUnlock() + } + }() + + deleteDone := make(chan struct{}) + go func() { + nc.DeleteGroup([]uint64{10}) + close(deleteDone) + }() + + deadline := time.Now().Add(2 * time.Second) + for nc.listMu.TryRLock() { + nc.listMu.RUnlock() + if time.Now().After(deadline) { + t.Fatal("DeleteGroup did not queue for listMu") + } + time.Sleep(time.Millisecond) + } + + updateDone := make(chan struct{}) + go func() { + nc.UpdateGroup(&model.NotificationGroup{ + Common: model.Common{ID: 10, UserID: 2}, + Name: "group", + }, nil) + close(updateDone) + }() + + deadline = time.Now().Add(2 * time.Second) + for nc.groupMu.TryLock() { + nc.groupMu.Unlock() + if time.Now().After(deadline) { + t.Fatal("neither group mutation acquired groupMu") + } + time.Sleep(time.Millisecond) + } + + nc.listMu.RUnlock() + readLockHeld = false + + timer := time.NewTimer(2 * time.Second) + defer timer.Stop() + for deleteDone != nil || updateDone != nil { + select { + case <-deleteDone: + deleteDone = nil + case <-updateDone: + updateDone = nil + case <-timer.C: + t.Fatal("concurrent UpdateGroup and DeleteGroup deadlocked") + } + } +}