From e23bc48281c384faf2b1d62b2b80f6aaa2aea364 Mon Sep 17 00:00:00 2001 From: avasconcelos114 Date: Fri, 14 Aug 2026 12:55:20 +0300 Subject: [PATCH 1/5] MM-67925 Added deduplication of DM notifications --- server/webhook.go | 49 ++++++++++++- server/webhook_test.go | 153 ++++++++++++++++++++++++++++++++++++++++- 2 files changed, 198 insertions(+), 4 deletions(-) diff --git a/server/webhook.go b/server/webhook.go index c18e8f6f..e71504bc 100644 --- a/server/webhook.go +++ b/server/webhook.go @@ -5,6 +5,8 @@ package main import ( "context" + "crypto/sha256" + "encoding/hex" "fmt" "io" "net/http" @@ -18,11 +20,18 @@ import ( "github.com/mattermost/mattermost-plugin-gitlab/server/webhook" "github.com/mattermost/mattermost/server/public/model" + "github.com/mattermost/mattermost/server/public/pluginapi" ) const ( webhookTimeout = 10 * time.Second eventSourceParentPipeline = "parent_pipeline" + + // notificationDedupTTL is the window during which a duplicate DM for the + // same recipient and message is suppressed (e.g. duplicate webhook + // deliveries from overlapping group/project hooks or GitLab retries). + notificationDedupTTL = 30 * time.Second + notificationDedupKeyFmt = "notif_dedup_%s" ) type gitlabRetreiver struct { @@ -184,9 +193,7 @@ func (p *Plugin) handleWebhook(w http.ResponseWriter, r *http.Request) { continue } if info.Settings.Notifications { - if err := p.CreateBotDMPost(userTo, res.Message, "custom_git_review_request"); err != nil { - p.client.Log.Warn("can't send dm post", "err", err.Error()) - } + p.sendDMNotification(userTo, res.Message) } } } @@ -206,6 +213,42 @@ func (p *Plugin) handleWebhook(w http.ResponseWriter, r *http.Request) { } } +// notificationDedupKey returns a KV key that uniquely identifies a DM +// notification by its recipient and rendered message, so that repeated +// deliveries of the same notification collapse onto the same key. +func notificationDedupKey(recipientID, message string) string { + hash := sha256.Sum256([]byte(recipientID + "_" + message)) + return fmt.Sprintf(notificationDedupKeyFmt, hex.EncodeToString(hash[:])) +} + +// sendDMNotification sends a bot DM to userID, deduplicating against +// duplicate webhook deliveries (e.g. overlapping group/project hooks, GitLab +// retries, or concurrent delivery across cluster nodes). It atomically claims +// a short-lived KV key before posting, so only the first delivery to claim +// the key actually sends the DM. +func (p *Plugin) sendDMNotification(userID, message string) { + dedupKey := notificationDedupKey(userID, message) + claimed, kvErr := p.client.KV.Set(dedupKey, true, + pluginapi.SetExpiry(notificationDedupTTL), + pluginapi.SetAtomic(nil), + ) + if kvErr != nil { + // Fail open: send the notification rather than dropping it. + p.client.Log.Warn("failed to claim notification dedup key, sending DM anyway", "err", kvErr.Error()) + } else if !claimed { + // Another delivery already claimed this notification. + p.client.Log.Debug("notification already claimed, skipping", "dedup_key", dedupKey) + return + } + + if err := p.CreateBotDMPost(userID, message, "custom_git_review_request"); err != nil { + // Keep the claim: CreateBotDMPost may have persisted the post despite + // returning an error, so releasing it would let a retry post a + // duplicate. Let the claim TTL expire on its own. + p.client.Log.Warn("can't send dm post", "err", err.Error()) + } +} + func (p *Plugin) sendRefreshIfNotAlreadySent(alreadySentRefresh map[string]bool, gitlabUsername string) string { if len(gitlabUsername) == 0 || alreadySentRefresh[gitlabUsername] { return "" diff --git a/server/webhook_test.go b/server/webhook_test.go index a695d56a..daec707b 100644 --- a/server/webhook_test.go +++ b/server/webhook_test.go @@ -6,16 +6,23 @@ package main import ( "bytes" "context" + "encoding/json" "net/http" "net/http/httptest" + "strings" + "sync" + "sync/atomic" "testing" "github.com/mattermost/mattermost/server/public/model" "github.com/mattermost/mattermost/server/public/plugin/plugintest" "github.com/mattermost/mattermost/server/public/pluginapi" "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/mock" + "github.com/stretchr/testify/require" gitlabLib "github.com/xanzy/go-gitlab" + "github.com/mattermost/mattermost-plugin-gitlab/server/gitlab" "github.com/mattermost/mattermost-plugin-gitlab/server/webhook" ) @@ -38,7 +45,11 @@ func (fakeWebhookHandler) HandleMergeRequest(_ context.Context, _ *gitlabLib.Mer } func (fakeWebhookHandler) HandleIssueComment(_ context.Context, _ *gitlabLib.IssueCommentEvent) ([]*webhook.HandleWebhook, []string, error) { - return nil, []string{}, nil + return []*webhook.HandleWebhook{{ + Message: "hello", + From: "test", + ToUsers: []string{"known"}, + }}, []string{}, nil } func (fakeWebhookHandler) HandleMergeRequestComment(_ context.Context, _ *gitlabLib.MergeCommentEvent) ([]*webhook.HandleWebhook, []string, error) { @@ -191,3 +202,143 @@ func TestHandleWebhookForChildPipelineNotficationEnabled(t *testing.T) { mock.AssertCalled(t, "PublishWebSocketEvent", WsEventRefresh, map[string]any(nil), &model.WebsocketBroadcast{UserId: "1"}) mock.AssertNumberOfCalls(t, "PublishWebSocketEvent", 1) } + +func TestNotificationDedupKey(t *testing.T) { + t.Run("same recipient and message produce the same key", func(t *testing.T) { + assert.Equal(t, + notificationDedupKey("user-1", "hello"), + notificationDedupKey("user-1", "hello")) + }) + + t.Run("different recipients produce different keys", func(t *testing.T) { + assert.NotEqual(t, + notificationDedupKey("user-1", "hello"), + notificationDedupKey("user-2", "hello")) + }) + + t.Run("different messages produce different keys", func(t *testing.T) { + assert.NotEqual(t, + notificationDedupKey("user-1", "hello"), + notificationDedupKey("user-1", "goodbye")) + }) + + t.Run("key has the expected prefix and length", func(t *testing.T) { + key := notificationDedupKey("user-1", "hello") + assert.True(t, strings.HasPrefix(key, "notif_dedup_")) + assert.Len(t, key, len("notif_dedup_")+64) // sha256 hex digest is 64 chars + }) +} + +// noteIssueCommentBody is a minimal GitLab "Note Hook" payload that parses +// into an *gitlabLib.IssueCommentEvent. +const noteIssueCommentBody = `{"object_kind":"note","user":{"username":"test"},"object_attributes":{"noteable_type":"Issue"}}` //nolint:misspell // "noteable_type" is GitLab's actual webhook field name + +func newIssueCommentWebhookRequest() *http.Request { + req := httptest.NewRequest("POST", "/", bytes.NewBufferString(noteIssueCommentBody)) + req.Header.Add("X-Gitlab-Token", "secret") + req.Header.Add("X-Gitlab-Event", string(gitlabLib.EventTypeNote)) + return req +} + +// setupDedupTestPlugin wires up a Plugin and mock API for a single known +// recipient ("known" GitLab username, mapped to mattermostUserID) with +// notifications enabled, ready to receive the "hello" DM from +// fakeWebhookHandler.HandleIssueComment. +func setupDedupTestPlugin(t *testing.T) (*Plugin, *plugintest.API) { + t.Helper() + + const mattermostUserID = "1" + + p := &Plugin{configuration: &configuration{WebhookSecret: "secret"}, WebhookHandler: fakeWebhookHandler{}} + + userInfo := &gitlab.UserInfo{ + UserID: mattermostUserID, + Settings: &gitlab.UserSettings{Notifications: true}, + } + infoJSON, err := json.Marshal(userInfo) + require.NoError(t, err) + + api := &plugintest.API{} + api.On("KVGet", "test_gitlabusername").Return(nil, nil) + api.On("KVGet", "known_gitlabusername").Return([]byte(mattermostUserID), nil) + api.On("KVGet", mattermostUserID+GitlabUserInfoKey).Return(infoJSON, nil) + api.On("PublishWebSocketEvent", WsEventRefresh, mock.Anything, mock.Anything).Return(nil) + api.On("LogInfo", "new msg", "message", "hello", "from", "test").Return(nil) + api.On("LogDebug", "notification already claimed, skipping", "dedup_key", mock.Anything).Return(nil) + api.On("GetDirectChannel", mattermostUserID, p.BotUserID).Return(&model.Channel{Id: "dm-channel"}, nil) + api.On("CreatePost", mock.MatchedBy(func(post *model.Post) bool { + return post.ChannelId == "dm-channel" && post.Message == "hello" + })).Return(&model.Post{}, nil) + + p.SetAPI(api) + p.client = pluginapi.NewClient(api, p.Driver) + + return p, api +} + +func TestHandleWebhookDeduplicatesDuplicateDelivery(t *testing.T) { + p, api := setupDedupTestPlugin(t) + + var claims atomic.Int32 + api.On("KVSetWithOptions", mock.AnythingOfType("string"), mock.Anything, mock.Anything).Return( + func(string, []byte, model.PluginKVSetOptions) (bool, *model.AppError) { + return claims.Add(1) == 1, nil + }, + ) + + // Simulate the same GitLab event being delivered twice, e.g. via + // overlapping group/project hooks or a GitLab retry. + for range 2 { + w := httptest.NewRecorder() + p.handleWebhook(w, newIssueCommentWebhookRequest()) + assert.Equal(t, http.StatusOK, w.Result().StatusCode) + } + + api.AssertNumberOfCalls(t, "CreatePost", 1) +} + +func TestHandleWebhookDeduplicatesConcurrentDeliveries(t *testing.T) { + p, api := setupDedupTestPlugin(t) + + var claims atomic.Int32 + api.On("KVSetWithOptions", mock.AnythingOfType("string"), mock.Anything, mock.Anything).Return( + func(string, []byte, model.PluginKVSetOptions) (bool, *model.AppError) { + return claims.Add(1) == 1, nil + }, + ) + + const deliveries = 25 + var wg sync.WaitGroup + wg.Add(deliveries) + start := make(chan struct{}) + for range deliveries { + go func() { + defer wg.Done() + <-start + w := httptest.NewRecorder() + p.handleWebhook(w, newIssueCommentWebhookRequest()) + }() + } + close(start) + wg.Wait() + + api.AssertNumberOfCalls(t, "CreatePost", 1) +} + +func TestSendDMNotificationFailsOpenOnKVError(t *testing.T) { + p := &Plugin{} + + api := &plugintest.API{} + api.On("KVSetWithOptions", mock.AnythingOfType("string"), mock.Anything, mock.Anything). + Return(false, model.NewAppError("KVSetWithOptions", "id", nil, "boom", http.StatusInternalServerError)) + api.On("GetDirectChannel", "user-1", p.BotUserID).Return(&model.Channel{Id: "dm-channel"}, nil) + api.On("CreatePost", mock.Anything).Return(&model.Post{}, nil) + api.On("LogWarn", mock.Anything, mock.Anything, mock.Anything).Return(nil) + + p.SetAPI(api) + p.client = pluginapi.NewClient(api, p.Driver) + + p.sendDMNotification("user-1", "hello") + + api.AssertNumberOfCalls(t, "CreatePost", 1) +} From 21b5489ad26788fbca745b4e6ff8415d8f8e19e2 Mon Sep 17 00:00:00 2001 From: avasconcelos114 Date: Fri, 14 Aug 2026 13:32:26 +0300 Subject: [PATCH 2/5] Releasing claim when DM fails to give retry a chance to succeed --- server/plugin.go | 7 +++++- server/webhook.go | 10 ++++++--- server/webhook_test.go | 48 ++++++++++++++++++++++++++++++++++++++++++ 3 files changed, 61 insertions(+), 4 deletions(-) diff --git a/server/plugin.go b/server/plugin.go index 1723b92c..877974dd 100644 --- a/server/plugin.go +++ b/server/plugin.go @@ -610,11 +610,16 @@ func (p *Plugin) registerChimeraURL() { p.chimeraURL = os.Getenv("MM_PLUGINSETTINGS_CHIMERAOAUTHPROXYURL") } +// errDMChannelUnavailable marks a CreateBotDMPost failure that happened +// before any post was attempted (i.e. resolving the bot's DM channel), so +// callers can be sure nothing was persisted. +var errDMChannelUnavailable = errors.New("bot DM channel unavailable") + func (p *Plugin) CreateBotDMPost(userID, message, postType string) error { channel, err := p.client.Channel.GetDirect(userID, p.BotUserID) if err != nil { p.client.Log.Warn("Couldn't get bot's DM channel", "user_id", userID) - return err + return fmt.Errorf("%w: %w", errDMChannelUnavailable, err) } post := &model.Post{ diff --git a/server/webhook.go b/server/webhook.go index e71504bc..c6a0b436 100644 --- a/server/webhook.go +++ b/server/webhook.go @@ -7,6 +7,7 @@ import ( "context" "crypto/sha256" "encoding/hex" + "errors" "fmt" "io" "net/http" @@ -242,10 +243,13 @@ func (p *Plugin) sendDMNotification(userID, message string) { } if err := p.CreateBotDMPost(userID, message, "custom_git_review_request"); err != nil { - // Keep the claim: CreateBotDMPost may have persisted the post despite - // returning an error, so releasing it would let a retry post a - // duplicate. Let the claim TTL expire on its own. p.client.Log.Warn("can't send dm post", "err", err.Error()) + if errors.Is(err, errDMChannelUnavailable) { + if delErr := p.client.KV.Delete(dedupKey); delErr != nil { + p.client.Log.Warn("failed to release notification dedup key", "err", delErr.Error()) + } + return + } } } diff --git a/server/webhook_test.go b/server/webhook_test.go index daec707b..56395ece 100644 --- a/server/webhook_test.go +++ b/server/webhook_test.go @@ -342,3 +342,51 @@ func TestSendDMNotificationFailsOpenOnKVError(t *testing.T) { api.AssertNumberOfCalls(t, "CreatePost", 1) } + +// TestSendDMNotificationReleasesClaimOnDirectChannelFailure verifies that a +// failure resolving the bot's DM channel (before any post is attempted) +// releases the dedup claim, so a subsequent retry isn't suppressed for the +// rest of the TTL. +func TestSendDMNotificationReleasesClaimOnDirectChannelFailure(t *testing.T) { + p := &Plugin{} + dedupKey := notificationDedupKey("user-1", "hello") + + api := &plugintest.API{} + api.On("KVSetWithOptions", dedupKey, []byte("true"), mock.Anything).Return(true, nil).Once() + api.On("GetDirectChannel", "user-1", p.BotUserID). + Return(nil, model.NewAppError("GetDirectChannel", "id", nil, "boom", http.StatusInternalServerError)) + api.On("KVSetWithOptions", dedupKey, isNilBytes, mock.Anything).Return(true, nil).Once() + api.On("LogWarn", mock.Anything, mock.Anything, mock.Anything).Return(nil) + + p.SetAPI(api) + p.client = pluginapi.NewClient(api, p.Driver) + + p.sendDMNotification("user-1", "hello") + + api.AssertNotCalled(t, "CreatePost", mock.Anything) + api.AssertCalled(t, "KVSetWithOptions", dedupKey, isNilBytes, mock.Anything) +} + +// TestSendDMNotificationKeepsClaimOnCreatePostFailure verifies that a +// CreatePost failure (which may have persisted the post despite the error) +// does not release the dedup claim, so a retry can't post a duplicate. +func TestSendDMNotificationKeepsClaimOnCreatePostFailure(t *testing.T) { + p := &Plugin{} + dedupKey := notificationDedupKey("user-1", "hello") + + api := &plugintest.API{} + api.On("KVSetWithOptions", dedupKey, []byte("true"), mock.Anything).Return(true, nil).Once() + api.On("GetDirectChannel", "user-1", p.BotUserID).Return(&model.Channel{Id: "dm-channel"}, nil) + api.On("CreatePost", mock.Anything). + Return(nil, model.NewAppError("CreatePost", "id", nil, "boom", http.StatusInternalServerError)) + api.On("LogWarn", mock.Anything, mock.Anything, mock.Anything).Return(nil) + // CreateBotDMPost logs its own "CreatePost failed" warning with 3 key/value pairs. + api.On("LogWarn", mock.Anything, mock.Anything, mock.Anything, mock.Anything, mock.Anything, mock.Anything, mock.Anything).Return(nil) + + p.SetAPI(api) + p.client = pluginapi.NewClient(api, p.Driver) + + p.sendDMNotification("user-1", "hello") + + api.AssertNotCalled(t, "KVSetWithOptions", dedupKey, isNilBytes, mock.Anything) +} From 90bcddf036088863f0ba7c01300cdd681caa51ab Mon Sep 17 00:00:00 2001 From: avasconcelos114 Date: Fri, 14 Aug 2026 13:39:35 +0300 Subject: [PATCH 3/5] Fixing conditions for early claim release --- server/webhook.go | 4 +++- server/webhook_test.go | 24 ++++++++++++++++++++++++ 2 files changed, 27 insertions(+), 1 deletion(-) diff --git a/server/webhook.go b/server/webhook.go index c6a0b436..eda7cd30 100644 --- a/server/webhook.go +++ b/server/webhook.go @@ -244,7 +244,9 @@ func (p *Plugin) sendDMNotification(userID, message string) { if err := p.CreateBotDMPost(userID, message, "custom_git_review_request"); err != nil { p.client.Log.Warn("can't send dm post", "err", err.Error()) - if errors.Is(err, errDMChannelUnavailable) { + // Only release a claim we actually own: if the KV.Set above failed + // (fail-open path), no claim was ever written, so there's nothing to delete. + if errors.Is(err, errDMChannelUnavailable) && kvErr == nil && claimed { if delErr := p.client.KV.Delete(dedupKey); delErr != nil { p.client.Log.Warn("failed to release notification dedup key", "err", delErr.Error()) } diff --git a/server/webhook_test.go b/server/webhook_test.go index 56395ece..8375c495 100644 --- a/server/webhook_test.go +++ b/server/webhook_test.go @@ -390,3 +390,27 @@ func TestSendDMNotificationKeepsClaimOnCreatePostFailure(t *testing.T) { api.AssertNotCalled(t, "KVSetWithOptions", dedupKey, isNilBytes, mock.Anything) } + +// TestSendDMNotificationSkipsDeleteWhenClaimWasNeverMade verifies that when +// the initial KV.Set claim fails (fail-open path) and CreateBotDMPost then +// fails with a DM-channel lookup error, no KV.Delete is attempted, since no +// claim was ever written for this recipient/message. +func TestSendDMNotificationSkipsDeleteWhenClaimWasNeverMade(t *testing.T) { + p := &Plugin{} + dedupKey := notificationDedupKey("user-1", "hello") + + api := &plugintest.API{} + api.On("KVSetWithOptions", dedupKey, []byte("true"), mock.Anything). + Return(false, model.NewAppError("KVSetWithOptions", "id", nil, "boom", http.StatusInternalServerError)).Once() + api.On("GetDirectChannel", "user-1", p.BotUserID). + Return(nil, model.NewAppError("GetDirectChannel", "id", nil, "boom", http.StatusInternalServerError)) + api.On("LogWarn", mock.Anything, mock.Anything, mock.Anything).Return(nil) + + p.SetAPI(api) + p.client = pluginapi.NewClient(api, p.Driver) + + p.sendDMNotification("user-1", "hello") + + api.AssertNotCalled(t, "KVSetWithOptions", dedupKey, isNilBytes, mock.Anything) + api.AssertNumberOfCalls(t, "KVSetWithOptions", 1) +} From d00db57be3d79dd55ba3b4f551fdd95edde262fb Mon Sep 17 00:00:00 2001 From: avasconcelos114 Date: Wed, 19 Aug 2026 10:13:07 +0300 Subject: [PATCH 4/5] Applying PR feedback --- server/webhook.go | 121 +++++++++++----- server/webhook_test.go | 319 ++++++++++++++++++++++++++++++++--------- 2 files changed, 332 insertions(+), 108 deletions(-) diff --git a/server/webhook.go b/server/webhook.go index eda7cd30..81ad2cf1 100644 --- a/server/webhook.go +++ b/server/webhook.go @@ -6,9 +6,11 @@ package main import ( "context" "crypto/sha256" + "encoding/binary" "encoding/hex" "errors" "fmt" + "hash" "io" "net/http" "time" @@ -28,11 +30,9 @@ const ( webhookTimeout = 10 * time.Second eventSourceParentPipeline = "parent_pipeline" - // notificationDedupTTL is the window during which a duplicate DM for the - // same recipient and message is suppressed (e.g. duplicate webhook - // deliveries from overlapping group/project hooks or GitLab retries). - notificationDedupTTL = 30 * time.Second + webhookDedupTTL = 30 * time.Second notificationDedupKeyFmt = "notif_dedup_%s" + channelPostDedupKeyFmt = "chan_dedup_%s" ) type gitlabRetreiver struct { @@ -200,61 +200,104 @@ func (p *Plugin) handleWebhook(w http.ResponseWriter, r *http.Request) { } for _, to := range res.ToChannels { if len(res.Message) > 0 { - post := &model.Post{ - UserId: p.BotUserID, - Message: res.Message, - ChannelId: to, - } - if err := p.client.Post.CreatePost(post); err != nil { - p.client.Log.Warn("can't create post for webhook event", "err", err.Error()) - } + p.sendChannelNotification(to, res.Message) } } p.sendRefreshIfNotAlreadySent(alreadySentRefresh, res.From) } } -// notificationDedupKey returns a KV key that uniquely identifies a DM -// notification by its recipient and rendered message, so that repeated -// deliveries of the same notification collapse onto the same key. +// writeDedupToken length-prefixes s so distinct token sequences can't hash to +// the same key (e.g. "a_b"+"c" versus "a"+"b_c"). +func writeDedupToken(h hash.Hash, s string) { + var length [8]byte + binary.BigEndian.PutUint64(length[:], uint64(len(s))) + // hash.Hash.Write never returns an error. + _, _ = h.Write(length[:]) + _, _ = h.Write([]byte(s)) +} + func notificationDedupKey(recipientID, message string) string { - hash := sha256.Sum256([]byte(recipientID + "_" + message)) - return fmt.Sprintf(notificationDedupKeyFmt, hex.EncodeToString(hash[:])) + h := sha256.New() + for _, token := range []string{recipientID, message} { + writeDedupToken(h, token) + } + return fmt.Sprintf(notificationDedupKeyFmt, hex.EncodeToString(h.Sum(nil))) } -// sendDMNotification sends a bot DM to userID, deduplicating against -// duplicate webhook deliveries (e.g. overlapping group/project hooks, GitLab -// retries, or concurrent delivery across cluster nodes). It atomically claims -// a short-lived KV key before posting, so only the first delivery to claim -// the key actually sends the DM. -func (p *Plugin) sendDMNotification(userID, message string) { - dedupKey := notificationDedupKey(userID, message) - claimed, kvErr := p.client.KV.Set(dedupKey, true, - pluginapi.SetExpiry(notificationDedupTTL), +func channelPostDedupKey(channelID, message string) string { + h := sha256.New() + for _, token := range []string{channelID, message} { + writeDedupToken(h, token) + } + return fmt.Sprintf(channelPostDedupKeyFmt, hex.EncodeToString(h.Sum(nil))) +} + +// claimDedupKey atomically claims dedupKey: SetAtomic(nil) only writes when the +// key is absent, so concurrent deliveries can't both win. A KV error fails open, +// returning deliver without owned since nothing was written. +func (p *Plugin) claimDedupKey(caller, dedupKey string) (deliver, owned bool) { + claimed, err := p.client.KV.Set(dedupKey, true, + pluginapi.SetExpiry(webhookDedupTTL), pluginapi.SetAtomic(nil), ) - if kvErr != nil { - // Fail open: send the notification rather than dropping it. - p.client.Log.Warn("failed to claim notification dedup key, sending DM anyway", "err", kvErr.Error()) - } else if !claimed { - // Another delivery already claimed this notification. - p.client.Log.Debug("notification already claimed, skipping", "dedup_key", dedupKey) + switch { + case err != nil: + p.client.Log.Warn(caller+": failed to claim dedup key, delivering anyway", "dedup_key", dedupKey, "err", err.Error()) + return true, false + case !claimed: + p.client.Log.Debug(caller+": another delivery already claimed this notification, skipping", "dedup_key", dedupKey) + return false, false + } + return true, true +} + +// releaseDedupKey lets a redelivery retry instead of waiting out the TTL. Only +// call it when this delivery owns the claim and nothing was posted. +func (p *Plugin) releaseDedupKey(caller, dedupKey string) { + if err := p.client.KV.Delete(dedupKey); err != nil { + p.client.Log.Warn(caller+": failed to release dedup key; redeliveries of this event will be skipped until the claim expires", + "dedup_key", dedupKey, "err", err.Error()) + } +} + +func (p *Plugin) sendDMNotification(userID, message string) { + dedupKey := notificationDedupKey(userID, message) + deliver, owned := p.claimDedupKey("sendDMNotification", dedupKey) + if !deliver { return } if err := p.CreateBotDMPost(userID, message, "custom_git_review_request"); err != nil { p.client.Log.Warn("can't send dm post", "err", err.Error()) - // Only release a claim we actually own: if the KV.Set above failed - // (fail-open path), no claim was ever written, so there's nothing to delete. - if errors.Is(err, errDMChannelUnavailable) && kvErr == nil && claimed { - if delErr := p.client.KV.Delete(dedupKey); delErr != nil { - p.client.Log.Warn("failed to release notification dedup key", "err", delErr.Error()) - } - return + // CreatePost may have persisted the DM despite erroring, so only the + // pre-post lookup failure is safe to release. + if owned && errors.Is(err, errDMChannelUnavailable) { + p.releaseDedupKey("sendDMNotification", dedupKey) } } } +// sendChannelNotification keys the claim on channel and message rather than on +// the subscription, so overlapping subscriptions collapse into one post. +func (p *Plugin) sendChannelNotification(channelID, message string) { + dedupKey := channelPostDedupKey(channelID, message) + if deliver, _ := p.claimDedupKey("sendChannelNotification", dedupKey); !deliver { + return + } + + post := &model.Post{ + UserId: p.BotUserID, + Message: message, + ChannelId: channelID, + } + // The claim is deliberately kept on failure: CreatePost may have persisted + // the post, and releasing would let a redelivery duplicate it. + if err := p.client.Post.CreatePost(post); err != nil { + p.client.Log.Warn("can't create post for webhook event", "err", err.Error()) + } +} + func (p *Plugin) sendRefreshIfNotAlreadySent(alreadySentRefresh map[string]bool, gitlabUsername string) string { if len(gitlabUsername) == 0 || alreadySentRefresh[gitlabUsername] { return "" diff --git a/server/webhook_test.go b/server/webhook_test.go index 8375c495..fa9f881a 100644 --- a/server/webhook_test.go +++ b/server/webhook_test.go @@ -11,8 +11,8 @@ import ( "net/http/httptest" "strings" "sync" - "sync/atomic" "testing" + "time" "github.com/mattermost/mattermost/server/public/model" "github.com/mattermost/mattermost/server/public/plugin/plugintest" @@ -142,6 +142,7 @@ func TestHandleWebhookToChannel(t *testing.T) { mock.On("LogInfo", "new msg", "message", "hello", "from", "test").Return(nil) mock.On("LogInfo", "userFrom", "from", "1").Return(nil) mock.On("CreatePost", &model.Post{Id: "", CreateAt: 0, UpdateAt: 0, EditAt: 0, DeleteAt: 0, IsPinned: false, UserId: "", ChannelId: "town-square", RootId: "", OriginalId: "", Message: "hello", MessageSource: "", Type: "", Hashtags: "", Filenames: model.StringArray(nil), FileIds: model.StringArray(nil), PendingPostId: "", HasReactions: false, Metadata: (*model.PostMetadata)(nil)}).Return(&model.Post{}, nil) + fakeDedupKV(mock) p.SetAPI(mock) p.client = pluginapi.NewClient(mock, p.Driver) @@ -227,10 +228,109 @@ func TestNotificationDedupKey(t *testing.T) { assert.True(t, strings.HasPrefix(key, "notif_dedup_")) assert.Len(t, key, len("notif_dedup_")+64) // sha256 hex digest is 64 chars }) + + // Without the length prefix these concatenate to the same bytes. + t.Run("shifting the recipient/message boundary does not collide", func(t *testing.T) { + assert.NotEqual(t, + notificationDedupKey("user-1", "hello"), + notificationDedupKey("user-1h", "ello")) + }) +} + +func TestChannelPostDedupKey(t *testing.T) { + t.Run("same channel and message produce the same key", func(t *testing.T) { + assert.Equal(t, + channelPostDedupKey("channel-1", "hello"), + channelPostDedupKey("channel-1", "hello")) + }) + + t.Run("different channels produce different keys", func(t *testing.T) { + assert.NotEqual(t, + channelPostDedupKey("channel-1", "hello"), + channelPostDedupKey("channel-2", "hello")) + }) + + t.Run("different messages produce different keys", func(t *testing.T) { + assert.NotEqual(t, + channelPostDedupKey("channel-1", "hello"), + channelPostDedupKey("channel-1", "goodbye")) + }) + + t.Run("key has the expected prefix and length", func(t *testing.T) { + key := channelPostDedupKey("channel-1", "hello") + assert.True(t, strings.HasPrefix(key, "chan_dedup_")) + assert.Len(t, key, len("chan_dedup_")+64) // sha256 hex digest is 64 chars + }) + + // Without the length prefix these concatenate to the same bytes. + t.Run("shifting the channel/message boundary does not collide", func(t *testing.T) { + assert.NotEqual(t, + channelPostDedupKey("channel-1", "hello"), + channelPostDedupKey("channel-1h", "ello")) + }) + + t.Run("channel and DM keys never collide", func(t *testing.T) { + assert.NotEqual(t, + channelPostDedupKey("id-1", "hello"), + notificationDedupKey("id-1", "hello")) + }) +} + +func isDedupClaimOptions(opts model.PluginKVSetOptions) bool { + return opts.Atomic && opts.OldValue == nil && + opts.ExpireInSeconds == int64(webhookDedupTTL/time.Second) +} + +// isDedupReleaseOptions matches KV.Delete, which pluginapi implements as a +// plain non-atomic write of a nil value. +func isDedupReleaseOptions(opts model.PluginKVSetOptions) bool { + return !opts.Atomic && opts.ExpireInSeconds == 0 +} + +// fakeDedupKV emulates the KV store's atomic-claim semantics keyed on the real +// dedup key, and returns an accessor for the currently held claims. +func fakeDedupKV(api *plugintest.API) func() []string { + var mu sync.Mutex + claimed := map[string]bool{} + + api.On("KVSetWithOptions", mock.AnythingOfType("string"), mock.Anything, mock.MatchedBy(isDedupClaimOptions)). + Return(func(key string, _ []byte, _ model.PluginKVSetOptions) (bool, *model.AppError) { + mu.Lock() + defer mu.Unlock() + if claimed[key] { + return false, nil + } + claimed[key] = true + return true, nil + }) + + api.On("KVSetWithOptions", mock.AnythingOfType("string"), isNilBytes, mock.MatchedBy(isDedupReleaseOptions)). + Return(func(key string, _ []byte, _ model.PluginKVSetOptions) (bool, *model.AppError) { + mu.Lock() + defer mu.Unlock() + delete(claimed, key) + return true, nil + }).Maybe() + + return func() []string { + mu.Lock() + defer mu.Unlock() + keys := make([]string, 0, len(claimed)) + for k := range claimed { + keys = append(keys, k) + } + return keys + } +} + +func newTestPluginWithAPI(api *plugintest.API) *Plugin { + p := &Plugin{} + p.SetAPI(api) + p.client = pluginapi.NewClient(api, p.Driver) + return p } -// noteIssueCommentBody is a minimal GitLab "Note Hook" payload that parses -// into an *gitlabLib.IssueCommentEvent. +// Minimal GitLab "Note Hook" payload that parses into an IssueCommentEvent. const noteIssueCommentBody = `{"object_kind":"note","user":{"username":"test"},"object_attributes":{"noteable_type":"Issue"}}` //nolint:misspell // "noteable_type" is GitLab's actual webhook field name func newIssueCommentWebhookRequest() *http.Request { @@ -240,10 +340,8 @@ func newIssueCommentWebhookRequest() *http.Request { return req } -// setupDedupTestPlugin wires up a Plugin and mock API for a single known -// recipient ("known" GitLab username, mapped to mattermostUserID) with -// notifications enabled, ready to receive the "hello" DM from -// fakeWebhookHandler.HandleIssueComment. +// setupDedupTestPlugin readies a plugin for the "hello" DM that +// fakeWebhookHandler.HandleIssueComment sends to the "known" user. func setupDedupTestPlugin(t *testing.T) (*Plugin, *plugintest.API) { t.Helper() @@ -264,7 +362,7 @@ func setupDedupTestPlugin(t *testing.T) (*Plugin, *plugintest.API) { api.On("KVGet", mattermostUserID+GitlabUserInfoKey).Return(infoJSON, nil) api.On("PublishWebSocketEvent", WsEventRefresh, mock.Anything, mock.Anything).Return(nil) api.On("LogInfo", "new msg", "message", "hello", "from", "test").Return(nil) - api.On("LogDebug", "notification already claimed, skipping", "dedup_key", mock.Anything).Return(nil) + api.On("LogDebug", mock.Anything, "dedup_key", mock.Anything).Return(nil) api.On("GetDirectChannel", mattermostUserID, p.BotUserID).Return(&model.Channel{Id: "dm-channel"}, nil) api.On("CreatePost", mock.MatchedBy(func(post *model.Post) bool { return post.ChannelId == "dm-channel" && post.Message == "hello" @@ -278,16 +376,8 @@ func setupDedupTestPlugin(t *testing.T) (*Plugin, *plugintest.API) { func TestHandleWebhookDeduplicatesDuplicateDelivery(t *testing.T) { p, api := setupDedupTestPlugin(t) + claimedKeys := fakeDedupKV(api) - var claims atomic.Int32 - api.On("KVSetWithOptions", mock.AnythingOfType("string"), mock.Anything, mock.Anything).Return( - func(string, []byte, model.PluginKVSetOptions) (bool, *model.AppError) { - return claims.Add(1) == 1, nil - }, - ) - - // Simulate the same GitLab event being delivered twice, e.g. via - // overlapping group/project hooks or a GitLab retry. for range 2 { w := httptest.NewRecorder() p.handleWebhook(w, newIssueCommentWebhookRequest()) @@ -295,17 +385,12 @@ func TestHandleWebhookDeduplicatesDuplicateDelivery(t *testing.T) { } api.AssertNumberOfCalls(t, "CreatePost", 1) + assert.Equal(t, []string{notificationDedupKey("1", "hello")}, claimedKeys()) } func TestHandleWebhookDeduplicatesConcurrentDeliveries(t *testing.T) { p, api := setupDedupTestPlugin(t) - - var claims atomic.Int32 - api.On("KVSetWithOptions", mock.AnythingOfType("string"), mock.Anything, mock.Anything).Return( - func(string, []byte, model.PluginKVSetOptions) (bool, *model.AppError) { - return claims.Add(1) == 1, nil - }, - ) + claimedKeys := fakeDedupKV(api) const deliveries = 25 var wg sync.WaitGroup @@ -323,94 +408,190 @@ func TestHandleWebhookDeduplicatesConcurrentDeliveries(t *testing.T) { wg.Wait() api.AssertNumberOfCalls(t, "CreatePost", 1) + assert.Equal(t, []string{notificationDedupKey("1", "hello")}, claimedKeys()) } func TestSendDMNotificationFailsOpenOnKVError(t *testing.T) { - p := &Plugin{} - api := &plugintest.API{} - api.On("KVSetWithOptions", mock.AnythingOfType("string"), mock.Anything, mock.Anything). + api.On("KVSetWithOptions", mock.AnythingOfType("string"), mock.Anything, mock.MatchedBy(isDedupClaimOptions)). Return(false, model.NewAppError("KVSetWithOptions", "id", nil, "boom", http.StatusInternalServerError)) - api.On("GetDirectChannel", "user-1", p.BotUserID).Return(&model.Channel{Id: "dm-channel"}, nil) api.On("CreatePost", mock.Anything).Return(&model.Post{}, nil) - api.On("LogWarn", mock.Anything, mock.Anything, mock.Anything).Return(nil) + api.On("LogWarn", mock.Anything, mock.Anything, mock.Anything, mock.Anything, mock.Anything).Return(nil) - p.SetAPI(api) - p.client = pluginapi.NewClient(api, p.Driver) + p := newTestPluginWithAPI(api) + api.On("GetDirectChannel", "user-1", p.BotUserID).Return(&model.Channel{Id: "dm-channel"}, nil) p.sendDMNotification("user-1", "hello") api.AssertNumberOfCalls(t, "CreatePost", 1) } -// TestSendDMNotificationReleasesClaimOnDirectChannelFailure verifies that a -// failure resolving the bot's DM channel (before any post is attempted) -// releases the dedup claim, so a subsequent retry isn't suppressed for the -// rest of the TTL. func TestSendDMNotificationReleasesClaimOnDirectChannelFailure(t *testing.T) { - p := &Plugin{} - dedupKey := notificationDedupKey("user-1", "hello") - api := &plugintest.API{} - api.On("KVSetWithOptions", dedupKey, []byte("true"), mock.Anything).Return(true, nil).Once() - api.On("GetDirectChannel", "user-1", p.BotUserID). - Return(nil, model.NewAppError("GetDirectChannel", "id", nil, "boom", http.StatusInternalServerError)) - api.On("KVSetWithOptions", dedupKey, isNilBytes, mock.Anything).Return(true, nil).Once() + claimedKeys := fakeDedupKV(api) api.On("LogWarn", mock.Anything, mock.Anything, mock.Anything).Return(nil) + api.On("LogWarn", mock.Anything, mock.Anything, mock.Anything, mock.Anything, mock.Anything).Return(nil) - p.SetAPI(api) - p.client = pluginapi.NewClient(api, p.Driver) + p := newTestPluginWithAPI(api) + api.On("GetDirectChannel", "user-1", p.BotUserID). + Return(nil, model.NewAppError("GetDirectChannel", "id", nil, "boom", http.StatusInternalServerError)) p.sendDMNotification("user-1", "hello") api.AssertNotCalled(t, "CreatePost", mock.Anything) - api.AssertCalled(t, "KVSetWithOptions", dedupKey, isNilBytes, mock.Anything) + assert.Empty(t, claimedKeys(), "the claim should be released when no post was attempted") } -// TestSendDMNotificationKeepsClaimOnCreatePostFailure verifies that a -// CreatePost failure (which may have persisted the post despite the error) -// does not release the dedup claim, so a retry can't post a duplicate. func TestSendDMNotificationKeepsClaimOnCreatePostFailure(t *testing.T) { - p := &Plugin{} - dedupKey := notificationDedupKey("user-1", "hello") - api := &plugintest.API{} - api.On("KVSetWithOptions", dedupKey, []byte("true"), mock.Anything).Return(true, nil).Once() - api.On("GetDirectChannel", "user-1", p.BotUserID).Return(&model.Channel{Id: "dm-channel"}, nil) + claimedKeys := fakeDedupKV(api) api.On("CreatePost", mock.Anything). Return(nil, model.NewAppError("CreatePost", "id", nil, "boom", http.StatusInternalServerError)) api.On("LogWarn", mock.Anything, mock.Anything, mock.Anything).Return(nil) - // CreateBotDMPost logs its own "CreatePost failed" warning with 3 key/value pairs. + // CreateBotDMPost logs its own warning with 3 key/value pairs. api.On("LogWarn", mock.Anything, mock.Anything, mock.Anything, mock.Anything, mock.Anything, mock.Anything, mock.Anything).Return(nil) - p.SetAPI(api) - p.client = pluginapi.NewClient(api, p.Driver) + p := newTestPluginWithAPI(api) + api.On("GetDirectChannel", "user-1", p.BotUserID).Return(&model.Channel{Id: "dm-channel"}, nil) p.sendDMNotification("user-1", "hello") - api.AssertNotCalled(t, "KVSetWithOptions", dedupKey, isNilBytes, mock.Anything) + assert.Equal(t, []string{notificationDedupKey("user-1", "hello")}, claimedKeys()) } -// TestSendDMNotificationSkipsDeleteWhenClaimWasNeverMade verifies that when -// the initial KV.Set claim fails (fail-open path) and CreateBotDMPost then -// fails with a DM-channel lookup error, no KV.Delete is attempted, since no -// claim was ever written for this recipient/message. func TestSendDMNotificationSkipsDeleteWhenClaimWasNeverMade(t *testing.T) { - p := &Plugin{} - dedupKey := notificationDedupKey("user-1", "hello") - api := &plugintest.API{} - api.On("KVSetWithOptions", dedupKey, []byte("true"), mock.Anything). + api.On("KVSetWithOptions", mock.AnythingOfType("string"), mock.Anything, mock.MatchedBy(isDedupClaimOptions)). Return(false, model.NewAppError("KVSetWithOptions", "id", nil, "boom", http.StatusInternalServerError)).Once() + api.On("LogWarn", mock.Anything, mock.Anything, mock.Anything).Return(nil) + api.On("LogWarn", mock.Anything, mock.Anything, mock.Anything, mock.Anything, mock.Anything).Return(nil) + + p := newTestPluginWithAPI(api) api.On("GetDirectChannel", "user-1", p.BotUserID). Return(nil, model.NewAppError("GetDirectChannel", "id", nil, "boom", http.StatusInternalServerError)) - api.On("LogWarn", mock.Anything, mock.Anything, mock.Anything).Return(nil) + p.sendDMNotification("user-1", "hello") + + api.AssertNotCalled(t, "KVSetWithOptions", mock.Anything, isNilBytes, mock.Anything) + api.AssertNumberOfCalls(t, "KVSetWithOptions", 1) +} + +// setupChannelDedupTestPlugin readies a plugin for the "hello" post that +// fakeWebhookHandler.HandleMergeRequest sends to "town-square". +func setupChannelDedupTestPlugin(t *testing.T) (*Plugin, *plugintest.API) { + t.Helper() + + api := &plugintest.API{} + api.On("KVGet", "test_gitlabusername").Return(nil, nil) + api.On("PublishWebSocketEvent", WsEventRefresh, mock.Anything, mock.Anything).Return(nil) + api.On("LogInfo", "new msg", "message", "hello", "from", "test").Return(nil) + api.On("LogDebug", mock.Anything, "dedup_key", mock.Anything).Return(nil) + api.On("CreatePost", mock.MatchedBy(func(post *model.Post) bool { + return post.ChannelId == "town-square" && post.Message == "hello" + })).Return(&model.Post{}, nil) + + p := &Plugin{configuration: &configuration{WebhookSecret: "secret"}, WebhookHandler: fakeWebhookHandler{}} p.SetAPI(api) p.client = pluginapi.NewClient(api, p.Driver) - p.sendDMNotification("user-1", "hello") + return p, api +} - api.AssertNotCalled(t, "KVSetWithOptions", dedupKey, isNilBytes, mock.Anything) - api.AssertNumberOfCalls(t, "KVSetWithOptions", 1) +func newMergeRequestWebhookRequest() *http.Request { + req := httptest.NewRequest("POST", "/", bytes.NewBufferString(`{"user": {"username":"test"}}`)) + req.Header.Add("X-Gitlab-Token", "secret") + req.Header.Add("X-Gitlab-Event", string(gitlabLib.EventTypeMergeRequest)) + return req +} + +func TestHandleWebhookDeduplicatesDuplicateChannelDelivery(t *testing.T) { + p, api := setupChannelDedupTestPlugin(t) + claimedKeys := fakeDedupKV(api) + + for range 2 { + w := httptest.NewRecorder() + p.handleWebhook(w, newMergeRequestWebhookRequest()) + assert.Equal(t, http.StatusOK, w.Result().StatusCode) + } + + api.AssertNumberOfCalls(t, "CreatePost", 1) + assert.Equal(t, []string{channelPostDedupKey("town-square", "hello")}, claimedKeys()) +} + +func TestHandleWebhookDeduplicatesConcurrentChannelDeliveries(t *testing.T) { + p, api := setupChannelDedupTestPlugin(t) + claimedKeys := fakeDedupKV(api) + + const deliveries = 25 + var wg sync.WaitGroup + wg.Add(deliveries) + start := make(chan struct{}) + for range deliveries { + go func() { + defer wg.Done() + <-start + w := httptest.NewRecorder() + p.handleWebhook(w, newMergeRequestWebhookRequest()) + }() + } + close(start) + wg.Wait() + + api.AssertNumberOfCalls(t, "CreatePost", 1) + assert.Equal(t, []string{channelPostDedupKey("town-square", "hello")}, claimedKeys()) +} + +func TestSendChannelNotification(t *testing.T) { + t.Run("different channels each get their own post", func(t *testing.T) { + api := &plugintest.API{} + claimedKeys := fakeDedupKV(api) + api.On("CreatePost", mock.Anything).Return(&model.Post{}, nil) + + p := newTestPluginWithAPI(api) + p.sendChannelNotification("channel-1", "hello") + p.sendChannelNotification("channel-2", "hello") + + // Two keys means the channel ID is part of the dedup identity. + api.AssertNumberOfCalls(t, "CreatePost", 2) + assert.Len(t, claimedKeys(), 2) + }) + + t.Run("different messages to one channel each get their own post", func(t *testing.T) { + api := &plugintest.API{} + claimedKeys := fakeDedupKV(api) + api.On("CreatePost", mock.Anything).Return(&model.Post{}, nil) + + p := newTestPluginWithAPI(api) + p.sendChannelNotification("channel-1", "hello") + p.sendChannelNotification("channel-1", "goodbye") + + api.AssertNumberOfCalls(t, "CreatePost", 2) + assert.Len(t, claimedKeys(), 2) + }) + + t.Run("fails open when the KV claim errors", func(t *testing.T) { + api := &plugintest.API{} + api.On("KVSetWithOptions", mock.AnythingOfType("string"), mock.Anything, mock.MatchedBy(isDedupClaimOptions)). + Return(false, model.NewAppError("KVSetWithOptions", "id", nil, "boom", http.StatusInternalServerError)) + api.On("CreatePost", mock.Anything).Return(&model.Post{}, nil) + api.On("LogWarn", mock.Anything, mock.Anything, mock.Anything, mock.Anything, mock.Anything).Return(nil) + + p := newTestPluginWithAPI(api) + p.sendChannelNotification("channel-1", "hello") + + api.AssertNumberOfCalls(t, "CreatePost", 1) + }) + + t.Run("keeps the claim when the post fails", func(t *testing.T) { + api := &plugintest.API{} + claimedKeys := fakeDedupKV(api) + api.On("CreatePost", mock.Anything). + Return(nil, model.NewAppError("CreatePost", "id", nil, "boom", http.StatusInternalServerError)) + api.On("LogWarn", mock.Anything, mock.Anything, mock.Anything).Return(nil) + + p := newTestPluginWithAPI(api) + p.sendChannelNotification("channel-1", "hello") + + assert.Equal(t, []string{channelPostDedupKey("channel-1", "hello")}, claimedKeys()) + }) } From b2ac0911f3d56d1ee69a4ee2ea158e3c8a2796e6 Mon Sep 17 00:00:00 2001 From: avasconcelos114 Date: Wed, 19 Aug 2026 14:07:56 +0300 Subject: [PATCH 5/5] Applying PR feedback --- server/webhook.go | 39 +++++++----- server/webhook_test.go | 132 +++++++++++++++++++++++++++-------------- 2 files changed, 110 insertions(+), 61 deletions(-) diff --git a/server/webhook.go b/server/webhook.go index 81ad2cf1..d67be60b 100644 --- a/server/webhook.go +++ b/server/webhook.go @@ -234,36 +234,45 @@ func channelPostDedupKey(channelID, message string) string { } // claimDedupKey atomically claims dedupKey: SetAtomic(nil) only writes when the -// key is absent, so concurrent deliveries can't both win. A KV error fails open, -// returning deliver without owned since nothing was written. -func (p *Plugin) claimDedupKey(caller, dedupKey string) (deliver, owned bool) { - claimed, err := p.client.KV.Set(dedupKey, true, +// key is absent, so concurrent deliveries can't both win. It returns the token +// identifying this claim, which releaseDedupKey needs to prove ownership. A KV +// error fails open, returning deliver with an empty token since nothing was +// written. +func (p *Plugin) claimDedupKey(caller, dedupKey string) (deliver bool, claimToken string) { + token := model.NewId() + claimed, err := p.client.KV.Set(dedupKey, []byte(token), pluginapi.SetExpiry(webhookDedupTTL), pluginapi.SetAtomic(nil), ) switch { case err != nil: p.client.Log.Warn(caller+": failed to claim dedup key, delivering anyway", "dedup_key", dedupKey, "err", err.Error()) - return true, false + return true, "" case !claimed: p.client.Log.Debug(caller+": another delivery already claimed this notification, skipping", "dedup_key", dedupKey) - return false, false + return false, "" } - return true, true + return true, token } -// releaseDedupKey lets a redelivery retry instead of waiting out the TTL. Only -// call it when this delivery owns the claim and nothing was posted. -func (p *Plugin) releaseDedupKey(caller, dedupKey string) { - if err := p.client.KV.Delete(dedupKey); err != nil { +// releaseDedupKey lets a redelivery retry instead of waiting out the TTL. It +// deletes only a claim still matching claimToken: a delivery slow enough for its +// claim to expire must not evict the claim of whichever delivery succeeded it. +func (p *Plugin) releaseDedupKey(caller, dedupKey, claimToken string) { + released, appErr := p.API.KVCompareAndDelete(dedupKey, []byte(claimToken)) + if appErr != nil { p.client.Log.Warn(caller+": failed to release dedup key; redeliveries of this event will be skipped until the claim expires", - "dedup_key", dedupKey, "err", err.Error()) + "dedup_key", dedupKey, "err", appErr.Error()) + return + } + if !released { + p.client.Log.Debug(caller+": dedup claim already expired and was reacquired, leaving it in place", "dedup_key", dedupKey) } } func (p *Plugin) sendDMNotification(userID, message string) { dedupKey := notificationDedupKey(userID, message) - deliver, owned := p.claimDedupKey("sendDMNotification", dedupKey) + deliver, claimToken := p.claimDedupKey("sendDMNotification", dedupKey) if !deliver { return } @@ -272,8 +281,8 @@ func (p *Plugin) sendDMNotification(userID, message string) { p.client.Log.Warn("can't send dm post", "err", err.Error()) // CreatePost may have persisted the DM despite erroring, so only the // pre-post lookup failure is safe to release. - if owned && errors.Is(err, errDMChannelUnavailable) { - p.releaseDedupKey("sendDMNotification", dedupKey) + if claimToken != "" && errors.Is(err, errDMChannelUnavailable) { + p.releaseDedupKey("sendDMNotification", dedupKey, claimToken) } } } diff --git a/server/webhook_test.go b/server/webhook_test.go index fa9f881a..7f56f530 100644 --- a/server/webhook_test.go +++ b/server/webhook_test.go @@ -281,46 +281,57 @@ func isDedupClaimOptions(opts model.PluginKVSetOptions) bool { opts.ExpireInSeconds == int64(webhookDedupTTL/time.Second) } -// isDedupReleaseOptions matches KV.Delete, which pluginapi implements as a -// plain non-atomic write of a nil value. -func isDedupReleaseOptions(opts model.PluginKVSetOptions) bool { - return !opts.Atomic && opts.ExpireInSeconds == 0 +// dedupKVFake emulates the KV store's atomic-claim and compare-and-delete +// semantics keyed on the real dedup key, so tests assert on the dedup identity +// itself rather than on call counts. +type dedupKVFake struct { + mu sync.Mutex + claimed map[string]string // dedup key -> claim token } -// fakeDedupKV emulates the KV store's atomic-claim semantics keyed on the real -// dedup key, and returns an accessor for the currently held claims. -func fakeDedupKV(api *plugintest.API) func() []string { - var mu sync.Mutex - claimed := map[string]bool{} +func fakeDedupKV(api *plugintest.API) *dedupKVFake { + kv := &dedupKVFake{claimed: map[string]string{}} api.On("KVSetWithOptions", mock.AnythingOfType("string"), mock.Anything, mock.MatchedBy(isDedupClaimOptions)). - Return(func(key string, _ []byte, _ model.PluginKVSetOptions) (bool, *model.AppError) { - mu.Lock() - defer mu.Unlock() - if claimed[key] { + Return(func(key string, value []byte, _ model.PluginKVSetOptions) (bool, *model.AppError) { + kv.mu.Lock() + defer kv.mu.Unlock() + if _, held := kv.claimed[key]; held { return false, nil } - claimed[key] = true + kv.claimed[key] = string(value) return true, nil }) - api.On("KVSetWithOptions", mock.AnythingOfType("string"), isNilBytes, mock.MatchedBy(isDedupReleaseOptions)). - Return(func(key string, _ []byte, _ model.PluginKVSetOptions) (bool, *model.AppError) { - mu.Lock() - defer mu.Unlock() - delete(claimed, key) + api.On("KVCompareAndDelete", mock.AnythingOfType("string"), mock.Anything). + Return(func(key string, oldValue []byte) (bool, *model.AppError) { + kv.mu.Lock() + defer kv.mu.Unlock() + if token, held := kv.claimed[key]; !held || token != string(oldValue) { + return false, nil + } + delete(kv.claimed, key) return true, nil }).Maybe() - return func() []string { - mu.Lock() - defer mu.Unlock() - keys := make([]string, 0, len(claimed)) - for k := range claimed { - keys = append(keys, k) - } - return keys + return kv +} + +func (kv *dedupKVFake) keys() []string { + kv.mu.Lock() + defer kv.mu.Unlock() + keys := make([]string, 0, len(kv.claimed)) + for k := range kv.claimed { + keys = append(keys, k) } + return keys +} + +// expire drops a claim the way the KV store's TTL would. +func (kv *dedupKVFake) expire(key string) { + kv.mu.Lock() + defer kv.mu.Unlock() + delete(kv.claimed, key) } func newTestPluginWithAPI(api *plugintest.API) *Plugin { @@ -376,7 +387,7 @@ func setupDedupTestPlugin(t *testing.T) (*Plugin, *plugintest.API) { func TestHandleWebhookDeduplicatesDuplicateDelivery(t *testing.T) { p, api := setupDedupTestPlugin(t) - claimedKeys := fakeDedupKV(api) + kv := fakeDedupKV(api) for range 2 { w := httptest.NewRecorder() @@ -385,12 +396,12 @@ func TestHandleWebhookDeduplicatesDuplicateDelivery(t *testing.T) { } api.AssertNumberOfCalls(t, "CreatePost", 1) - assert.Equal(t, []string{notificationDedupKey("1", "hello")}, claimedKeys()) + assert.Equal(t, []string{notificationDedupKey("1", "hello")}, kv.keys()) } func TestHandleWebhookDeduplicatesConcurrentDeliveries(t *testing.T) { p, api := setupDedupTestPlugin(t) - claimedKeys := fakeDedupKV(api) + kv := fakeDedupKV(api) const deliveries = 25 var wg sync.WaitGroup @@ -408,7 +419,7 @@ func TestHandleWebhookDeduplicatesConcurrentDeliveries(t *testing.T) { wg.Wait() api.AssertNumberOfCalls(t, "CreatePost", 1) - assert.Equal(t, []string{notificationDedupKey("1", "hello")}, claimedKeys()) + assert.Equal(t, []string{notificationDedupKey("1", "hello")}, kv.keys()) } func TestSendDMNotificationFailsOpenOnKVError(t *testing.T) { @@ -428,7 +439,7 @@ func TestSendDMNotificationFailsOpenOnKVError(t *testing.T) { func TestSendDMNotificationReleasesClaimOnDirectChannelFailure(t *testing.T) { api := &plugintest.API{} - claimedKeys := fakeDedupKV(api) + kv := fakeDedupKV(api) api.On("LogWarn", mock.Anything, mock.Anything, mock.Anything).Return(nil) api.On("LogWarn", mock.Anything, mock.Anything, mock.Anything, mock.Anything, mock.Anything).Return(nil) @@ -439,12 +450,12 @@ func TestSendDMNotificationReleasesClaimOnDirectChannelFailure(t *testing.T) { p.sendDMNotification("user-1", "hello") api.AssertNotCalled(t, "CreatePost", mock.Anything) - assert.Empty(t, claimedKeys(), "the claim should be released when no post was attempted") + assert.Empty(t, kv.keys(), "the claim should be released when no post was attempted") } func TestSendDMNotificationKeepsClaimOnCreatePostFailure(t *testing.T) { api := &plugintest.API{} - claimedKeys := fakeDedupKV(api) + kv := fakeDedupKV(api) api.On("CreatePost", mock.Anything). Return(nil, model.NewAppError("CreatePost", "id", nil, "boom", http.StatusInternalServerError)) api.On("LogWarn", mock.Anything, mock.Anything, mock.Anything).Return(nil) @@ -456,7 +467,7 @@ func TestSendDMNotificationKeepsClaimOnCreatePostFailure(t *testing.T) { p.sendDMNotification("user-1", "hello") - assert.Equal(t, []string{notificationDedupKey("user-1", "hello")}, claimedKeys()) + assert.Equal(t, []string{notificationDedupKey("user-1", "hello")}, kv.keys()) } func TestSendDMNotificationSkipsDeleteWhenClaimWasNeverMade(t *testing.T) { @@ -472,10 +483,39 @@ func TestSendDMNotificationSkipsDeleteWhenClaimWasNeverMade(t *testing.T) { p.sendDMNotification("user-1", "hello") - api.AssertNotCalled(t, "KVSetWithOptions", mock.Anything, isNilBytes, mock.Anything) + api.AssertNotCalled(t, "KVCompareAndDelete", mock.Anything, mock.Anything) api.AssertNumberOfCalls(t, "KVSetWithOptions", 1) } +// A delivery slow enough for its claim to expire must not evict the claim of +// whichever delivery reacquired the key, or the event posts twice. +func TestSendDMNotificationReleaseDoesNotEvictReacquiredClaim(t *testing.T) { + api := &plugintest.API{} + kv := fakeDedupKV(api) + api.On("LogDebug", mock.Anything, "dedup_key", mock.Anything).Return(nil) + + p := newTestPluginWithAPI(api) + dedupKey := notificationDedupKey("user-1", "hello") + + deliverA, tokenA := p.claimDedupKey("test", dedupKey) + require.True(t, deliverA) + require.NotEmpty(t, tokenA) + + // A stalls in GetDirect long enough for its claim to time out. + kv.expire(dedupKey) + + deliverB, tokenB := p.claimDedupKey("test", dedupKey) + require.True(t, deliverB) + require.NotEqual(t, tokenA, tokenB) + + // A finally fails and releases using its own, now-stale token. + p.releaseDedupKey("test", dedupKey, tokenA) + + require.Equal(t, []string{dedupKey}, kv.keys(), "B's claim must survive A's release") + deliverC, _ := p.claimDedupKey("test", dedupKey) + require.False(t, deliverC, "a later delivery must still be suppressed by B's claim") +} + // setupChannelDedupTestPlugin readies a plugin for the "hello" post that // fakeWebhookHandler.HandleMergeRequest sends to "town-square". func setupChannelDedupTestPlugin(t *testing.T) (*Plugin, *plugintest.API) { @@ -506,7 +546,7 @@ func newMergeRequestWebhookRequest() *http.Request { func TestHandleWebhookDeduplicatesDuplicateChannelDelivery(t *testing.T) { p, api := setupChannelDedupTestPlugin(t) - claimedKeys := fakeDedupKV(api) + kv := fakeDedupKV(api) for range 2 { w := httptest.NewRecorder() @@ -515,12 +555,12 @@ func TestHandleWebhookDeduplicatesDuplicateChannelDelivery(t *testing.T) { } api.AssertNumberOfCalls(t, "CreatePost", 1) - assert.Equal(t, []string{channelPostDedupKey("town-square", "hello")}, claimedKeys()) + assert.Equal(t, []string{channelPostDedupKey("town-square", "hello")}, kv.keys()) } func TestHandleWebhookDeduplicatesConcurrentChannelDeliveries(t *testing.T) { p, api := setupChannelDedupTestPlugin(t) - claimedKeys := fakeDedupKV(api) + kv := fakeDedupKV(api) const deliveries = 25 var wg sync.WaitGroup @@ -538,13 +578,13 @@ func TestHandleWebhookDeduplicatesConcurrentChannelDeliveries(t *testing.T) { wg.Wait() api.AssertNumberOfCalls(t, "CreatePost", 1) - assert.Equal(t, []string{channelPostDedupKey("town-square", "hello")}, claimedKeys()) + assert.Equal(t, []string{channelPostDedupKey("town-square", "hello")}, kv.keys()) } func TestSendChannelNotification(t *testing.T) { t.Run("different channels each get their own post", func(t *testing.T) { api := &plugintest.API{} - claimedKeys := fakeDedupKV(api) + kv := fakeDedupKV(api) api.On("CreatePost", mock.Anything).Return(&model.Post{}, nil) p := newTestPluginWithAPI(api) @@ -553,12 +593,12 @@ func TestSendChannelNotification(t *testing.T) { // Two keys means the channel ID is part of the dedup identity. api.AssertNumberOfCalls(t, "CreatePost", 2) - assert.Len(t, claimedKeys(), 2) + assert.Len(t, kv.keys(), 2) }) t.Run("different messages to one channel each get their own post", func(t *testing.T) { api := &plugintest.API{} - claimedKeys := fakeDedupKV(api) + kv := fakeDedupKV(api) api.On("CreatePost", mock.Anything).Return(&model.Post{}, nil) p := newTestPluginWithAPI(api) @@ -566,7 +606,7 @@ func TestSendChannelNotification(t *testing.T) { p.sendChannelNotification("channel-1", "goodbye") api.AssertNumberOfCalls(t, "CreatePost", 2) - assert.Len(t, claimedKeys(), 2) + assert.Len(t, kv.keys(), 2) }) t.Run("fails open when the KV claim errors", func(t *testing.T) { @@ -584,7 +624,7 @@ func TestSendChannelNotification(t *testing.T) { t.Run("keeps the claim when the post fails", func(t *testing.T) { api := &plugintest.API{} - claimedKeys := fakeDedupKV(api) + kv := fakeDedupKV(api) api.On("CreatePost", mock.Anything). Return(nil, model.NewAppError("CreatePost", "id", nil, "boom", http.StatusInternalServerError)) api.On("LogWarn", mock.Anything, mock.Anything, mock.Anything).Return(nil) @@ -592,6 +632,6 @@ func TestSendChannelNotification(t *testing.T) { p := newTestPluginWithAPI(api) p.sendChannelNotification("channel-1", "hello") - assert.Equal(t, []string{channelPostDedupKey("channel-1", "hello")}, claimedKeys()) + assert.Equal(t, []string{channelPostDedupKey("channel-1", "hello")}, kv.keys()) }) }