fix(notifications): avoid deadlock when creating issue notifications concurrently (#40166) (#40203)

Backport #40166 by @perfectra1n

Concurrent `notification-service` workers can deadlock in the
transaction for-loop.

The transaction is useless, so it is removed to avoid deadlock.

Co-authored-by: Jon Fuller <jonfuller2012@gmail.com>
Co-authored-by: wxiaoguang <wxiaoguang@gmail.com>

models/activities/notification.go

@@ -180,15 +180,14 @@ func updateIssueNotification(ctx context.Context, userID, issueID, commentID, up
 	}
 
 	// NOTICE: Only update comment id when the before notification on this issue is read, otherwise you may miss some old comments.
-	// But we need update update_by so that the notification will be reorder
-	var cols []string
+	// Always update so that updated_unix moves the notification to the top of the list
+	// Actually, it's unclear what UpdatedBy is used for, or maybe not used at all, or maybe it should just be dropped.
+	notification.UpdatedBy = updatedByID
+	cols := []string{"updated_by"}
 	if notification.Status == NotificationStatusRead {
 		notification.Status = NotificationStatusUnread
 		notification.CommentID = commentID
-		cols = []string{"status", "update_by", "comment_id"}
-	} else {
-		notification.UpdatedBy = updatedByID
-		cols = []string{"update_by"}
+		cols = append(cols, "status", "comment_id")
 	}
 
 	_, err = db.GetEngine(ctx).ID(notification.ID).Cols(cols...).Update(notification)

models/activities/notification_list.go

@@ -71,16 +71,6 @@ func (opts FindNotificationOptions) ToOrders() string {
 // receiverID > 0 just send to receiver, else send to all watcher
 // Returns the set of user IDs whose notification rows were created or updated.
 func CreateOrUpdateIssueNotifications(ctx context.Context, issueID, commentID, notificationAuthorID, receiverID int64) ([]int64, error) {
-	var notifiedIDs []int64
-	err := db.WithTx(ctx, func(ctx context.Context) error {
-		var innerErr error
-		notifiedIDs, innerErr = createOrUpdateIssueNotifications(ctx, issueID, commentID, notificationAuthorID, receiverID)
-		return innerErr
-	})
-	return notifiedIDs, err
-}
-
-func createOrUpdateIssueNotifications(ctx context.Context, issueID, commentID, notificationAuthorID, receiverID int64) ([]int64, error) {
 	// init
 	var toNotify container.Set[int64]
 	notifications, err := db.Find[Notification](ctx, FindNotificationOptions{
@@ -181,7 +171,11 @@ func createOrUpdateIssueNotifications(ctx context.Context, issueID, commentID, n
 		if notificationExists(notifications, issue.ID, userID) {
 			err = updateIssueNotification(ctx, userID, issue.ID, commentID, notificationAuthorID)
 		} else {
-			err = createIssueNotification(ctx, userID, issue, commentID, notificationAuthorID)
+			if err = createIssueNotification(ctx, userID, issue, commentID, notificationAuthorID); err != nil {
+				if errUpdate := updateIssueNotification(ctx, userID, issue.ID, commentID, notificationAuthorID); errUpdate == nil {
+					err = nil
+				}
+			}
 		}
 		if err != nil {
 			return nil, err

models/activities/notification_test.go

@@ -31,6 +31,18 @@ func TestCreateOrUpdateIssueNotifications(t *testing.T) {
 
 	notf = unittest.AssertExistsAndLoadBean(t, &activities_model.Notification{UserID: 4, IssueID: issue.ID})
 	assert.Equal(t, activities_model.NotificationStatusUnread, notf.Status)
+
+	// unread notification should get latest updated_unix when re-notified
+	_, err = db.GetEngine(t.Context()).ID(notf.ID).NoAutoTime().Cols("updated_unix").Update(&activities_model.Notification{UpdatedUnix: 123})
+	assert.NoError(t, err)
+	notf = unittest.AssertExistsAndLoadBean(t, &activities_model.Notification{ID: notf.ID})
+	assert.EqualValues(t, 123, notf.UpdatedUnix)
+
+	_, err = activities_model.CreateOrUpdateIssueNotifications(t.Context(), issue.ID, 0, 3, 0)
+	assert.NoError(t, err)
+	notf = unittest.AssertExistsAndLoadBean(t, &activities_model.Notification{ID: notf.ID})
+	assert.NotEqualValues(t, 123, notf.UpdatedUnix)
+	assert.EqualValues(t, 3, notf.UpdatedBy)
 }
 
 func TestCreateOrUpdateIssueNotificationsForAssigneeAndReviewer(t *testing.T) {