diff --git a/server/plugin/command_test.go b/server/plugin/command_test.go index f64f76b36..00166e56c 100644 --- a/server/plugin/command_test.go +++ b/server/plugin/command_test.go @@ -5,6 +5,7 @@ package plugin import ( "context" + "encoding/json" "fmt" "net/http" "net/http/httptest" @@ -1095,6 +1096,26 @@ func TestCreatePost(t *testing.T) { } } +// expectSubscriptionModify sets up the mock KV store so a single +// SetAtomicWithRetries call invokes the provided callback with the JSON +// encoding of initial as the old value, mirroring the real store's +// read-modify-write. It returns whatever error the callback returns. +func expectSubscriptionModify(mockKVStore *mocks.MockKvStore, initial *Subscriptions) { + mockKVStore.EXPECT().SetAtomicWithRetries(SubscriptionsKey, gomock.Any()).DoAndReturn( + func(_ string, valueFunc func(oldValue []byte) (any, error)) error { + var oldValue []byte + if initial != nil { + b, err := json.Marshal(initial) + if err != nil { + return err + } + oldValue = b + } + _, err := valueFunc(oldValue) + return err + }).Times(1) +} + func TestHandleUnsubscribe(t *testing.T) { mockKVStore, mockAPI, _, _, _ := GetTestSetup(t) p := getPluginTest(mockAPI, mockKVStore) @@ -1131,8 +1152,8 @@ func TestHandleUnsubscribe(t *testing.T) { name: "Failed to unsubscribe", parameters: []string{"owner/repo"}, setup: func() { - mockKVStore.EXPECT().Get(SubscriptionsKey, gomock.Any()).Return(errors.New("error occurred getting subscriptions")) - mockAPI.On("LogWarn", "Failed to unsubscribe", "repo", "repo", "error", "could not get subscriptions: could not get subscriptions from KVStore: error occurred getting subscriptions") + mockKVStore.EXPECT().SetAtomicWithRetries(SubscriptionsKey, gomock.Any()).Return(errors.New("error occurred setting subscriptions")).Times(1) + mockAPI.On("LogWarn", "Failed to unsubscribe", "repo", "repo", "error", "could not store subscriptions: error occurred setting subscriptions") }, assertions: func(result string) { assert.Equal(t, "Encountered an error trying to unsubscribe. Please try again.", result) @@ -1142,13 +1163,7 @@ func TestHandleUnsubscribe(t *testing.T) { name: "No subscription exists for repo in the channel", parameters: []string{"owner/repo"}, setup: func() { - mockKVStore.EXPECT().Get(SubscriptionsKey, gomock.Any()).DoAndReturn(func(key string, value **Subscriptions) error { - *value = &Subscriptions{Repositories: map[string][]*Subscription{}} - return nil - }).Times(1) - mockAPI.On("GetUser", MockUserID).Return(nil, &model.AppError{Message: "error getting user"}).Times(1) - mockAPI.On("LogWarn", "Error while fetching user details", "error", "error getting user").Times(1) - mockKVStore.EXPECT().SetAtomicWithRetries(SubscriptionsKey, gomock.Any()).Return(nil).Times(1) + expectSubscriptionModify(mockKVStore, &Subscriptions{Repositories: map[string][]*Subscription{}}) }, assertions: func(result string) { assert.Equal(t, "no subscription exists for `owner/repo` in the channel", result) @@ -1158,15 +1173,11 @@ func TestHandleUnsubscribe(t *testing.T) { name: "Error getting user details", parameters: []string{"owner/repo"}, setup: func() { - mockKVStore.EXPECT().Get(SubscriptionsKey, gomock.Any()).DoAndReturn(func(key string, value **Subscriptions) error { - *value = &Subscriptions{Repositories: map[string][]*Subscription{ - "owner/repo": {{ChannelID: MockChannelID, CreatorID: MockCreatorID, Repository: "owner/repo"}}, - }} - return nil - }).Times(1) + expectSubscriptionModify(mockKVStore, &Subscriptions{Repositories: map[string][]*Subscription{ + "owner/repo": {{ChannelID: MockChannelID, CreatorID: MockCreatorID, Repository: "owner/repo"}}, + }}) mockAPI.On("GetUser", MockUserID).Return(nil, &model.AppError{Message: "error getting user"}).Times(1) mockAPI.On("LogWarn", "Error while fetching user details", "error", "error getting user").Times(1) - mockKVStore.EXPECT().SetAtomicWithRetries(SubscriptionsKey, gomock.Any()).Return(nil).Times(1) }, assertions: func(result string) { assert.Equal(t, "error while fetching user details: error getting user", result) @@ -1176,17 +1187,13 @@ func TestHandleUnsubscribe(t *testing.T) { name: "Error creating post of unsubscribe with no repo", parameters: []string{"owner"}, setup: func() { - mockKVStore.EXPECT().Get(SubscriptionsKey, gomock.Any()).DoAndReturn(func(key string, value **Subscriptions) error { - *value = &Subscriptions{Repositories: map[string][]*Subscription{ - "owner/": {{ChannelID: MockChannelID, CreatorID: MockCreatorID, Repository: "owner"}}, - }} - return nil - }).Times(1) + expectSubscriptionModify(mockKVStore, &Subscriptions{Repositories: map[string][]*Subscription{ + "owner/": {{ChannelID: MockChannelID, CreatorID: MockCreatorID, Repository: "owner"}}, + }}) mockAPI.On("GetUser", MockUserID).Return(&model.User{Username: MockUsername}, nil).Times(1) mockAPI.On("CreatePost", mock.Anything).Return(nil, &model.AppError{Message: "error creating post"}).Times(1) post.Message = "@mockUsername unsubscribed this channel from [owner](https://github.com/owner)" mockAPI.On("LogWarn", "Error while creating post", "channel_id", mock.Anything, "error", "error creating post").Times(1) - mockKVStore.EXPECT().SetAtomicWithRetries(SubscriptionsKey, gomock.Any()).Return(nil).Times(1) }, assertions: func(result string) { assert.Equal(t, "@mockUsername unsubscribed this channel from [owner](https://github.com/owner) error creating the public post: error creating post", result) @@ -1196,15 +1203,11 @@ func TestHandleUnsubscribe(t *testing.T) { name: "Success unsubscribing with no repo", parameters: []string{"owner"}, setup: func() { - mockKVStore.EXPECT().Get(SubscriptionsKey, gomock.Any()).DoAndReturn(func(key string, value **Subscriptions) error { - *value = &Subscriptions{Repositories: map[string][]*Subscription{ - "owner/": {{ChannelID: MockChannelID, CreatorID: MockCreatorID, Repository: ""}}, - }} - return nil - }).Times(1) + expectSubscriptionModify(mockKVStore, &Subscriptions{Repositories: map[string][]*Subscription{ + "owner/": {{ChannelID: MockChannelID, CreatorID: MockCreatorID, Repository: ""}}, + }}) mockAPI.On("GetUser", MockUserID).Return(&model.User{Username: MockUsername}, nil).Times(1) mockAPI.On("CreatePost", mock.Anything).Return(post, nil).Times(1) - mockKVStore.EXPECT().SetAtomicWithRetries(SubscriptionsKey, gomock.Any()).Return(nil).Times(1) }, assertions: func(result string) { assert.Empty(t, result) @@ -1214,17 +1217,13 @@ func TestHandleUnsubscribe(t *testing.T) { name: "Error creating post of unsubscribe with no repo", parameters: []string{"owner/repo"}, setup: func() { - mockKVStore.EXPECT().Get(SubscriptionsKey, gomock.Any()).DoAndReturn(func(key string, value **Subscriptions) error { - *value = &Subscriptions{Repositories: map[string][]*Subscription{ - "owner/repo": {{ChannelID: MockChannelID, CreatorID: MockCreatorID, Repository: "owner/repo"}}, - }} - return nil - }).Times(1) + expectSubscriptionModify(mockKVStore, &Subscriptions{Repositories: map[string][]*Subscription{ + "owner/repo": {{ChannelID: MockChannelID, CreatorID: MockCreatorID, Repository: "owner/repo"}}, + }}) mockAPI.On("GetUser", MockUserID).Return(&model.User{Username: MockUsername}, nil).Times(1) mockAPI.On("CreatePost", mock.Anything).Return(nil, &model.AppError{Message: "error creating post"}).Times(1) post.Message = "@mockUsername Unsubscribed this channel from [owner/repo](https://github.com/owner/repo)\n Please delete the [webhook](https://github.com/owner/repo/settings/hooks) for this subscription unless it's required for other subscriptions." mockAPI.On("LogWarn", "Error while creating post", "channel_id", mock.Anything, "error", "error creating post").Times(1) - mockKVStore.EXPECT().SetAtomicWithRetries(SubscriptionsKey, gomock.Any()).Return(nil).Times(1) }, assertions: func(result string) { assert.Equal(t, "@mockUsername Unsubscribed this channel from [owner/repo](https://github.com/owner/repo)\n Please delete the [webhook](https://github.com/owner/repo/settings/hooks) for this subscription unless it's required for other subscriptions. error creating the public post: error creating post", result) @@ -1234,16 +1233,12 @@ func TestHandleUnsubscribe(t *testing.T) { name: "Success unsubscribing with repo", parameters: []string{"owner/repo"}, setup: func() { - mockKVStore.EXPECT().Get(SubscriptionsKey, gomock.Any()).DoAndReturn(func(key string, value **Subscriptions) error { - *value = &Subscriptions{Repositories: map[string][]*Subscription{ - "owner/repo": {{ChannelID: MockChannelID, CreatorID: MockCreatorID, Repository: "owner/repo"}}, - }} - return nil - }).Times(1) + expectSubscriptionModify(mockKVStore, &Subscriptions{Repositories: map[string][]*Subscription{ + "owner/repo": {{ChannelID: MockChannelID, CreatorID: MockCreatorID, Repository: "owner/repo"}}, + }}) mockAPI.ExpectedCalls = nil mockAPI.On("GetUser", MockUserID).Return(&model.User{Username: MockUsername}, nil).Times(1) mockAPI.On("CreatePost", mock.Anything).Return(post, nil).Times(1) - mockKVStore.EXPECT().SetAtomicWithRetries(SubscriptionsKey, gomock.Any()).Return(nil).Times(1) post.Message = "" }, assertions: func(result string) { diff --git a/server/plugin/subscription_race_test.go b/server/plugin/subscription_race_test.go new file mode 100644 index 000000000..506a7bc63 --- /dev/null +++ b/server/plugin/subscription_race_test.go @@ -0,0 +1,78 @@ +// Copyright (c) 2018-present Mattermost, Inc. All Rights Reserved. +// See LICENSE.txt for license information. + +package plugin + +import ( + "fmt" + "sync" + "testing" + + "github.com/mattermost/mattermost/server/public/pluginapi" +) + +// TestSubscriptionRace drives the plugin's real AddSubscription against the +// vendored pluginapi.MemoryStore, whose SetAtomicWithRetries is byte-identical +// to the production KVService. No plugin logic is mocked. Many channels +// subscribe to the same repository concurrently; afterwards we compare how many +// AddSubscription calls reported success (nil error) against how many +// subscriptions actually persisted. +// +// Correct behaviour: persisted == reported-success (every reported success is +// durable). Before the fix, the read-modify-write happened outside the atomic +// callback, so concurrent writers clobbered each other and persisted was far +// lower than reported-success (silent lost updates). +func TestSubscriptionRace(t *testing.T) { + const numChannels = 200 + + p := NewPlugin() + p.client = pluginapi.NewClient(p.API, p.Driver) + p.store = &pluginapi.MemoryStore{} + + var wg sync.WaitGroup + var mu sync.Mutex + reportedSuccess := 0 + + for i := 0; i < numChannels; i++ { + wg.Add(1) + go func(i int) { + defer wg.Done() + sub := &Subscription{ + ChannelID: fmt.Sprintf("channel-%d", i), + Repository: "org/repo", + } + if err := p.AddSubscription("org/repo", sub); err == nil { + mu.Lock() + reportedSuccess++ + mu.Unlock() + } + }(i) + } + wg.Wait() + + subs, err := p.GetSubscriptions() + if err != nil { + t.Fatalf("GetSubscriptions failed: %v", err) + } + persisted := 0 + for _, chans := range subs.Repositories { + persisted += len(chans) + } + + silentlyLost := reportedSuccess - persisted + t.Logf("concurrent AddSubscription calls : %d", numChannels) + t.Logf("AddSubscription returned success : %d", reportedSuccess) + t.Logf("subscriptions actually persisted : %d", persisted) + t.Logf("silently lost (success but gone) : %d", silentlyLost) + + if silentlyLost > 0 { + t.Fatalf("lost-update bug: %d subscriptions were reported as saved but silently dropped", silentlyLost) + } + + // Guard against a false pass: if every AddSubscription call failed, + // reportedSuccess and persisted are both 0, so silentlyLost is 0 and the + // check above passes without actually exercising the race. + if reportedSuccess == 0 { + t.Fatalf("no AddSubscription calls succeeded; the race was not exercised") + } +} diff --git a/server/plugin/subscriptions.go b/server/plugin/subscriptions.go index 5ae6dc592..33fe45fab 100644 --- a/server/plugin/subscriptions.go +++ b/server/plugin/subscriptions.go @@ -304,32 +304,28 @@ func (p *Plugin) GetSubscriptionsByChannel(channelID string) ([]*Subscription, e } func (p *Plugin) AddSubscription(repo string, sub *Subscription) error { - subs, err := p.GetSubscriptions() - if err != nil { - return errors.Wrap(err, "could not get subscriptions") - } + err := p.modifySubscriptions(func(subs *Subscriptions) error { + repoSubs := subs.Repositories[repo] + if repoSubs == nil { + repoSubs = []*Subscription{sub} + } else { + exists := false + for index, s := range repoSubs { + if s.ChannelID == sub.ChannelID { + repoSubs[index] = sub + exists = true + break + } + } - repoSubs := subs.Repositories[repo] - if repoSubs == nil { - repoSubs = []*Subscription{sub} - } else { - exists := false - for index, s := range repoSubs { - if s.ChannelID == sub.ChannelID { - repoSubs[index] = sub - exists = true - break + if !exists { + repoSubs = append(repoSubs, sub) } } - if !exists { - repoSubs = append(repoSubs, sub) - } - } - - subs.Repositories[repo] = repoSubs - - err = p.StoreSubscriptions(subs) + subs.Repositories[repo] = repoSubs + return nil + }) if err != nil { return errors.Wrap(err, "could not store subscriptions") } @@ -353,9 +349,28 @@ func (p *Plugin) GetSubscriptions() (*Subscriptions, error) { return subscriptions, nil } -func (p *Plugin) StoreSubscriptions(s *Subscriptions) error { - return p.store.SetAtomicWithRetries(SubscriptionsKey, func(_ []byte) (any, error) { - modifiedBytes, err := json.Marshal(s) +// modifySubscriptions performs an atomic read-modify-write of the whole +// subscriptions blob. The mutate function runs inside SetAtomicWithRetries' +// callback, so on every retry it receives the freshly re-read state and +// re-applies its change on top of it. This prevents concurrent mutations +// across channels from silently clobbering one another (lost updates). +func (p *Plugin) modifySubscriptions(mutate func(*Subscriptions) error) error { + return p.store.SetAtomicWithRetries(SubscriptionsKey, func(oldValue []byte) (any, error) { + subs := &Subscriptions{Repositories: map[string][]*Subscription{}} + if len(oldValue) > 0 { + if err := json.Unmarshal(oldValue, subs); err != nil { + return nil, errors.Wrap(err, "could not unmarshal subscriptions from KV store") + } + if subs.Repositories == nil { + subs.Repositories = map[string][]*Subscription{} + } + } + + if err := mutate(subs); err != nil { + return nil, err + } + + modifiedBytes, err := json.Marshal(subs) if err != nil { return nil, errors.Wrap(err, "could not store subscriptions in KV store") } @@ -419,34 +434,46 @@ func NewSubscriptionError(code int, err error) *SubscriptionError { return &SubscriptionError{Code: code, Error: err} } +// errStopModify is a sentinel returned from a modifySubscriptions mutate +// function to abort the atomic write without treating it as a store failure. +// SetAtomicWithRetries returns immediately (no retry) when the callback errors, +// so the caller inspects its own captured result instead of this error. +var errStopModify = errors.New("stop modifying subscriptions") + func (p *Plugin) Unsubscribe(channelID, repo, owner string) *SubscriptionError { repoWithOwner := fmt.Sprintf("%s/%s", owner, repo) - subs, err := p.GetSubscriptions() - if err != nil { - return NewSubscriptionError(InternalServerError, errors.Wrap(err, "could not get subscriptions")) - } + notFound := NewSubscriptionError(SubscriptionNotFound, errors.Errorf(SubscriptionUnavailable, strings.TrimSuffix(repoWithOwner, "/"))) - repoSubs := subs.Repositories[repoWithOwner] - if repoSubs == nil { - return NewSubscriptionError(SubscriptionNotFound, errors.Errorf(SubscriptionUnavailable, strings.TrimSuffix(repoWithOwner, "/"))) - } + var subErr *SubscriptionError + err := p.modifySubscriptions(func(subs *Subscriptions) error { + repoSubs := subs.Repositories[repoWithOwner] + if repoSubs == nil { + subErr = notFound + return errStopModify + } - removed := false - for index, sub := range repoSubs { - if sub.ChannelID == channelID { - repoSubs = append(repoSubs[:index], repoSubs[index+1:]...) - removed = true - break + removed := false + for index, sub := range repoSubs { + if sub.ChannelID == channelID { + repoSubs = append(repoSubs[:index], repoSubs[index+1:]...) + removed = true + break + } } - } - if !removed { - return NewSubscriptionError(SubscriptionNotFound, errors.Errorf(SubscriptionUnavailable, strings.TrimSuffix(repoWithOwner, "/"))) - } + if !removed { + subErr = notFound + return errStopModify + } - subs.Repositories[repoWithOwner] = repoSubs - if err := p.StoreSubscriptions(subs); err != nil { + subs.Repositories[repoWithOwner] = repoSubs + return nil + }) + if subErr != nil { + return subErr + } + if err != nil { return NewSubscriptionError(InternalServerError, errors.Wrap(err, "could not store subscriptions")) }