fix: harden notification cache synchronization

This commit is contained in:
naiba committed 2026-09-19 10:44:17 +00:00
1 parent 589f01f562
commit 9dab6a5039
4 files changed
+225 -55

No files matched your search

+6
View File
@@ -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
}
@@ -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)
}
}
+68 -55
View File
@@ -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,
+126
View File
@@ -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")
}
}
}