fix(vote): serialize vote count refreshes

internal/repo/activity/vote_repo.go

@@ -182,9 +182,67 @@ func (vr *VoteRepo) CancelVote(ctx context.Context, op *schema.VoteOperationInfo
 
 func (vr *VoteRepo) GetAndSaveVoteResult(ctx context.Context, objectID, objectType string) (
 	up, down int64, err error) {
-	up = vr.countVoteUp(ctx, objectID, objectType)
-	down = vr.countVoteDown(ctx, objectID, objectType)
-	err = vr.updateVotes(ctx, objectID, objectType, int(up-down))
+	var target any
+	switch objectType {
+	case constant.QuestionObjectType:
+		target = &entity.Question{}
+	case constant.AnswerObjectType:
+		target = &entity.Answer{}
+	case constant.CommentObjectType:
+		target = &entity.Comment{}
+	default:
+		return 0, 0, errors.BadRequest(reason.ObjectNotFound)
+	}
+
+	// Resolve configuration before opening the transaction: these lookups may
+	// need a separate connection, while SQLite uses a single connection.
+	upType, err := vr.activityRepo.GetActivityTypeByObjectType(ctx, objectType, constant.ActVoteUp)
+	if err != nil {
+		return 0, 0, err
+	}
+	activityTypes := []int{upType}
+	downType := 0
+	if objectType != constant.CommentObjectType {
+		downType, err = vr.activityRepo.GetActivityTypeByObjectType(ctx, objectType, constant.ActVoteDown)
+		if err != nil {
+			return 0, 0, err
+		}
+		activityTypes = append(activityTypes, downType)
+	}
+
+	_, err = vr.data.DB.Transaction(func(session *xorm.Session) (any, error) {
+		session = session.Context(ctx)
+		// Acquire the target's write lock before reading any vote totals.
+		// This also starts a write transaction on SQLite, where FOR UPDATE
+		// is unavailable. Keep the value and update timestamp unchanged.
+		if _, err := session.ID(objectID).NoAutoTime().Cols("vote_count").
+			SetExpr("vote_count", "vote_count").Update(target); err != nil {
+			return nil, err
+		}
+
+		var counts []struct {
+			ActivityType int
+			Count        int64 `xorm:"vote_count"`
+		}
+		if err := session.Table(new(entity.Activity)).Select("activity_type, COUNT(*) AS vote_count").
+			Where("object_id = ?", objectID).And("cancelled = ?", entity.ActivityAvailable).
+			In("activity_type", activityTypes).GroupBy("activity_type").Find(&counts); err != nil {
+			return nil, err
+		}
+		for _, count := range counts {
+			switch count.ActivityType {
+			case upType:
+				up = count.Count
+			case downType:
+				down = count.Count
+			}
+		}
+		_, err := session.ID(objectID).Cols("vote_count").SetExpr("vote_count", up-down).Update(target)
+		return nil, err
+	})
+	if err != nil {
+		return 0, 0, errors.InternalServer(reason.DatabaseError).WithError(err).WithStack()
+	}
 	return
 }
 
@@ -408,51 +466,6 @@ func (vr *VoteRepo) getExistActivity(ctx context.Context, op *schema.VoteOperati
 	return activities, nil
 }
 
-func (vr *VoteRepo) countVoteUp(ctx context.Context, objectID, objectType string) (count int64) {
-	count, err := vr.countVote(ctx, objectID, objectType, constant.ActVoteUp)
-	if err != nil {
-		log.Errorf("get vote up count error: %v", err)
-	}
-	return count
-}
-
-func (vr *VoteRepo) countVoteDown(ctx context.Context, objectID, objectType string) (count int64) {
-	count, err := vr.countVote(ctx, objectID, objectType, constant.ActVoteDown)
-	if err != nil {
-		log.Errorf("get vote down count error: %v", err)
-	}
-	return count
-}
-
-func (vr *VoteRepo) countVote(ctx context.Context, objectID, objectType, action string) (count int64, err error) {
-	activity := &entity.Activity{}
-	activityType, _ := vr.activityRepo.GetActivityTypeByObjectType(ctx, objectType, action)
-	count, err = vr.data.DB.Context(ctx).Where(builder.Eq{"object_id": objectID}).
-		And(builder.Eq{"activity_type": activityType}).
-		And(builder.Eq{"cancelled": 0}).
-		Count(activity)
-	if err != nil {
-		err = errors.InternalServer(reason.DatabaseError).WithError(err).WithStack()
-	}
-	return count, err
-}
-
-func (vr *VoteRepo) updateVotes(ctx context.Context, objectID, objectType string, voteCount int) (err error) {
-	session := vr.data.DB.Context(ctx)
-	switch objectType {
-	case constant.QuestionObjectType:
-		_, err = session.ID(objectID).Cols("vote_count").Update(&entity.Question{VoteCount: voteCount})
-	case constant.AnswerObjectType:
-		_, err = session.ID(objectID).Cols("vote_count").Update(&entity.Answer{VoteCount: voteCount})
-	case constant.CommentObjectType:
-		_, err = session.ID(objectID).Cols("vote_count").Update(&entity.Comment{VoteCount: voteCount})
-	}
-	if err != nil {
-		log.Error(err)
-	}
-	return
-}
-
 func (vr *VoteRepo) sendAchievementNotification(ctx context.Context, activityUserID, objectUserID, objectID string) {
 	objectType, err := obj.GetObjectTypeStrByObjectID(objectID)
 	if err != nil {

internal/repo/activity/vote_repo_test.go

@@ -0,0 +1,260 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *   http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package activity
+
+import (
+	"context"
+	"errors"
+	"path/filepath"
+	"strings"
+	"sync"
+	"testing"
+	"time"
+
+	"github.com/apache/answer/internal/base/constant"
+	"github.com/apache/answer/internal/base/data"
+	"github.com/apache/answer/internal/entity"
+	"github.com/apache/answer/internal/service/activity_common"
+	"github.com/stretchr/testify/require"
+	"xorm.io/xorm/contexts"
+	"xorm.io/xorm/schemas"
+)
+
+type voteResultActivityRepo struct {
+	activity_common.ActivityRepo
+	err error
+}
+
+func (r *voteResultActivityRepo) GetActivityTypeByObjectType(_ context.Context, objectType, action string) (int, error) {
+	if r.err != nil {
+		return 0, r.err
+	}
+	if action == constant.ActVoteUp {
+		return 1, nil
+	}
+	if objectType == constant.CommentObjectType {
+		return 0, errors.New("comments have no downvote activity type")
+	}
+	return 2, nil
+}
+
+type voteResultHook struct {
+	before func(*contexts.ContextHook) (context.Context, error)
+}
+
+func (h voteResultHook) BeforeProcess(c *contexts.ContextHook) (context.Context, error) {
+	return h.before(c)
+}
+
+func (h voteResultHook) AfterProcess(_ *contexts.ContextHook) error { return nil }
+
+func newVoteResultRepo(t *testing.T, connection string) *VoteRepo {
+	t.Helper()
+	db, err := data.NewDB(false, &data.Database{Driver: string(schemas.SQLITE), Connection: connection})
+	require.NoError(t, err)
+	t.Cleanup(func() { require.NoError(t, db.Close()) })
+	require.NoError(t, db.Sync2(&entity.Question{}, &entity.Answer{}, &entity.Comment{}, &entity.Activity{}))
+	return &VoteRepo{data: &data.Data{DB: db}, activityRepo: &voteResultActivityRepo{}}
+}
+
+func voteResultTarget(objectType string, count int) any {
+	switch objectType {
+	case constant.QuestionObjectType:
+		return &entity.Question{
+			ID: "10010000000000001", UserID: "101", LastEditUserID: "0",
+			AcceptedAnswerID: "0", LastAnswerID: "0", RevisionID: "0", VoteCount: count,
+		}
+	case constant.AnswerObjectType:
+		return &entity.Answer{
+			ID: "10010000000000001", UserID: "101", QuestionID: "0",
+			LastEditUserID: "0", RevisionID: "0", VoteCount: count,
+		}
+	default:
+		return &entity.Comment{
+			ID: "10010000000000001", UserID: "101", ObjectID: "0", QuestionID: "0", VoteCount: count,
+		}
+	}
+}
+
+func storedVoteResult(t *testing.T, repo *VoteRepo, objectType string) int64 {
+	t.Helper()
+	var result struct {
+		Count int64 `xorm:"vote_count"`
+	}
+	exist, err := repo.data.DB.Table(voteResultTarget(objectType, 0)).
+		Select("vote_count").Where("id = ?", "10010000000000001").Get(&result)
+	require.NoError(t, err)
+	require.True(t, exist)
+	return result.Count
+}
+
+func TestGetAndSaveVoteResultRejectsStaleTotals(t *testing.T) {
+	for _, objectType := range []string{constant.QuestionObjectType, constant.AnswerObjectType, constant.CommentObjectType} {
+		for _, tc := range []struct {
+			name        string
+			cancel      bool
+			newVoteType int
+			up, down    int64
+		}{
+			{"new upvote", false, 1, 2, 0},
+			{"cancel upvote", true, 0, 0, 0},
+			{"new downvote", false, 2, 1, 1},
+			{"switch to downvote", true, 2, 0, 1},
+		} {
+			if objectType == constant.CommentObjectType && tc.newVoteType == 2 {
+				continue
+			}
+			t.Run(objectType+"/"+tc.name, func(t *testing.T) {
+				connection := filepath.Join(t.TempDir(), "votes.db")
+				slow := newVoteResultRepo(t, connection)
+				fast := newVoteResultRepo(t, connection)
+				_, err := slow.data.DB.Insert(voteResultTarget(objectType, 0))
+				require.NoError(t, err)
+				firstVote := &entity.Activity{ObjectID: "10010000000000001", OriginalObjectID: "0", UserID: "101", ActivityType: 1}
+				_, err = slow.data.DB.Insert(firstVote)
+				require.NoError(t, err)
+
+				// Pause the older request just before its first target UPDATE.
+				// The original implementation has already calculated its stale
+				// counts here. A lock-before-count implementation has not.
+				paused, resume := make(chan struct{}), make(chan struct{})
+				unblock := sync.OnceFunc(func() { close(resume) })
+				defer unblock()
+				var once sync.Once
+				slow.data.DB.AddHook(voteResultHook{before: func(c *contexts.ContextHook) (context.Context, error) {
+					if strings.HasPrefix(strings.ToUpper(c.SQL), "UPDATE") && strings.Contains(c.SQL, "vote_count") {
+						once.Do(func() { close(paused); <-resume })
+					}
+					return c.Ctx, nil
+				}})
+				ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second)
+				defer cancel()
+				type result struct {
+					up, down int64
+					err      error
+				}
+				done := make(chan result, 1)
+				go func() {
+					up, down, err := slow.GetAndSaveVoteResult(ctx, "10010000000000001", objectType)
+					done <- result{up, down, err}
+				}()
+				select {
+				case <-paused:
+				case <-ctx.Done():
+					t.Fatal("older request did not reach the target update")
+				}
+
+				want := tc.up - tc.down
+				if tc.cancel {
+					_, err = fast.data.DB.ID(firstVote.ID).Cols("cancelled").
+						Update(&entity.Activity{Cancelled: entity.ActivityCancelled})
+					require.NoError(t, err)
+				}
+				if tc.newVoteType != 0 {
+					_, err = fast.data.DB.Insert(&entity.Activity{
+						ObjectID: "10010000000000001", OriginalObjectID: "0", UserID: "202", ActivityType: tc.newVoteType,
+					})
+					require.NoError(t, err)
+				}
+				_, _, err = fast.GetAndSaveVoteResult(ctx, "10010000000000001", objectType)
+				require.NoError(t, err)
+				require.Equal(t, want, storedVoteResult(t, fast, objectType))
+				unblock()
+				select {
+				case got := <-done:
+					require.NoError(t, got.err)
+					require.Equal(t, want, storedVoteResult(t, fast, objectType))
+					require.Equal(t, tc.up, got.up)
+					require.Equal(t, tc.down, got.down)
+				case <-ctx.Done():
+					t.Fatal("older request did not complete")
+				}
+			})
+		}
+	}
+}
+
+func TestGetAndSaveVoteResultCountsActiveVotes(t *testing.T) {
+	for _, objectType := range []string{constant.QuestionObjectType, constant.AnswerObjectType, constant.CommentObjectType} {
+		t.Run(objectType, func(t *testing.T) {
+			repo := newVoteResultRepo(t, filepath.Join(t.TempDir(), "votes.db"))
+			_, err := repo.data.DB.Insert(voteResultTarget(objectType, 9))
+			require.NoError(t, err)
+			votes := []*entity.Activity{
+				{ObjectID: "10010000000000001", OriginalObjectID: "0", UserID: "101", ActivityType: 1},
+				{ObjectID: "10010000000000001", OriginalObjectID: "0", UserID: "102", ActivityType: 1, Cancelled: entity.ActivityCancelled},
+				{ObjectID: "10010000000000001", OriginalObjectID: "0", UserID: "103", ActivityType: 7},
+				{ObjectID: "10010000000000002", OriginalObjectID: "0", UserID: "104", ActivityType: 1},
+			}
+			if objectType != constant.CommentObjectType {
+				votes = append(votes,
+					&entity.Activity{ObjectID: "10010000000000001", OriginalObjectID: "0", UserID: "105", ActivityType: 2},
+					&entity.Activity{ObjectID: "10010000000000001", OriginalObjectID: "0", UserID: "106", ActivityType: 2},
+				)
+			}
+			_, err = repo.data.DB.Insert(votes)
+			require.NoError(t, err)
+			up, down, err := repo.GetAndSaveVoteResult(context.Background(), "10010000000000001", objectType)
+			require.NoError(t, err)
+			require.Equal(t, int64(1), up)
+			if objectType == constant.CommentObjectType {
+				require.Zero(t, down)
+			} else {
+				require.Equal(t, int64(2), down)
+			}
+			require.Equal(t, up-down, storedVoteResult(t, repo, objectType))
+
+			_, err = repo.data.DB.Where("object_id = ?", "10010000000000001").
+				Cols("cancelled").Update(&entity.Activity{Cancelled: entity.ActivityCancelled})
+			require.NoError(t, err)
+			up, down, err = repo.GetAndSaveVoteResult(context.Background(), "10010000000000001", objectType)
+			require.NoError(t, err)
+			require.Zero(t, up)
+			require.Zero(t, down)
+			require.Zero(t, storedVoteResult(t, repo, objectType))
+		})
+	}
+}
+
+func TestGetAndSaveVoteResultDoesNotOverwriteOnCountError(t *testing.T) {
+	repo := newVoteResultRepo(t, filepath.Join(t.TempDir(), "votes.db"))
+	_, err := repo.data.DB.Insert(voteResultTarget(constant.QuestionObjectType, 7))
+	require.NoError(t, err)
+	repo.data.DB.AddHook(voteResultHook{before: func(c *contexts.ContextHook) (context.Context, error) {
+		if strings.Contains(strings.ToUpper(c.SQL), "COUNT(") {
+			return c.Ctx, errors.New("count query failed")
+		}
+		return c.Ctx, nil
+	}})
+	_, _, err = repo.GetAndSaveVoteResult(context.Background(), "10010000000000001", constant.QuestionObjectType)
+	require.Error(t, err)
+	require.Equal(t, int64(7), storedVoteResult(t, repo, constant.QuestionObjectType))
+}
+
+func TestGetAndSaveVoteResultDoesNotOverwriteOnActivityTypeError(t *testing.T) {
+	repo := newVoteResultRepo(t, filepath.Join(t.TempDir(), "votes.db"))
+	_, err := repo.data.DB.Insert(voteResultTarget(constant.QuestionObjectType, 7))
+	require.NoError(t, err)
+	lookupErr := errors.New("activity configuration unavailable")
+	repo.activityRepo = &voteResultActivityRepo{err: lookupErr}
+	_, _, err = repo.GetAndSaveVoteResult(context.Background(), "10010000000000001", constant.QuestionObjectType)
+	require.ErrorIs(t, err, lookupErr)
+	require.Equal(t, int64(7), storedVoteResult(t, repo, constant.QuestionObjectType))
+}