Skip to content

Fix lost-update race in channel subscriptions #1045

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We鈥檒l occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
aegonmyy wants to merge 2 commits into
base: master
Choose a base branch
from
Open

Fix lost-update race in channel subscriptions #1045

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
c3b8ae8
Fix lost-update race in channel subscriptions
aegonmyy Aug 2, 2026
235d14d
test: fail if no AddSubscription calls succeeded (avoid false pass)
aegonmyy Aug 4, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
83 changes: 39 additions & 44 deletions server/plugin/command_test.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -5,6 +5,7 @@ package plugin

import (
"context"
"encoding/json"
"fmt"
"net/http"
"net/http/httptest"
Expand Down Expand Up @@ -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)
Expand Down Expand Up @@ -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)
Expand All @@ -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)
Expand All @@ -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)
Expand All @@ -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)
Expand All @@ -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)
Expand All @@ -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)
Expand All @@ -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) {
Expand Down
78 changes: 78 additions & 0 deletions server/plugin/subscription_race_test.go
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -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)
}

@jgheithcock jgheithcock Aug 3, 2026

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

(Similar to mattermost/mattermost-plugin-gitlab#691.)
If all AddSubscription calls fail, subs will be empty and persisted will be 0 and so silentlyLost will be 0 and no failures will be reported. Suggest adding the following after silentlyLost check:

if reportedSuccess == 0 {
t.Fatalf("No AddSubscription calls succeeded")
}

@aegonmyy aegonmyy Aug 4, 2026

Copy link
Copy Markdown
Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Good catch, thanks. Added a guard after the silentlyLost check that fails the test if reportedSuccess == 0, so a run where every AddSubscription call failed (persisted 0, silentlyLost 0) no longer passes silently. Pushed in 235d14d.


// 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")
}
}
117 changes: 72 additions & 45 deletions server/plugin/subscriptions.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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")
}
Expand All @@ -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")
}
Expand Down Expand Up @@ -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"))
}

Expand Down