From fcc2d07383f25f7d6d29138b19571119d2080cfd Mon Sep 17 00:00:00 2001 From: avasconcelos114 Date: Fri, 14 Aug 2026 20:43:32 +0300 Subject: [PATCH 1/5] MM-70280 Applying dedup on channel subscription posts --- server/plugin_test.go | 1 + server/webhook.go | 46 +++++++++++-- server/webhook_http_test.go | 1 + server/webhook_parser_misc_test.go | 64 +++++++++++++++++ server/webhook_test.go | 106 +++++++++++++++++++++++++++++ 5 files changed, 211 insertions(+), 7 deletions(-) create mode 100644 server/webhook_test.go diff --git a/server/plugin_test.go b/server/plugin_test.go index e30c02b0..9c073922 100644 --- a/server/plugin_test.go +++ b/server/plugin_test.go @@ -127,6 +127,7 @@ func TestPlugin(t *testing.T) { api.On("LogWarn", mockAnythingOfTypeBatch("string", 13)...).Return(nil) api.On("KVGet", mock.AnythingOfType("string")).Return(make([]byte, 0), (*model.AppError)(nil)) + api.On("KVSetWithOptions", mock.AnythingOfType("string"), mock.Anything, mock.Anything).Return(true, (*model.AppError)(nil)) api.On("GetDirectChannel", mockAnythingOfTypeBatch("string", 2)...).Return( &model.Channel{}, (*model.AppError)(nil)) api.On("GetUserByUsername", "theuser").Return(&model.User{ diff --git a/server/webhook.go b/server/webhook.go index 388eed99..a8ef65c8 100644 --- a/server/webhook.go +++ b/server/webhook.go @@ -10,6 +10,7 @@ import ( "net/http" "net/url" "strconv" + "strings" "time" "github.com/pkg/errors" @@ -21,8 +22,9 @@ import ( ) const ( - notificationDedupTTL = 30 * time.Second + webhookDedupTTL = 30 * time.Second notificationDedupKeyFmt = "notif_dedup_%s" + channelPostDedupKeyFmt = "chan_dedup_%s" ) const ( @@ -78,8 +80,13 @@ func (wh webhook) PostToChannel(p *Plugin, instanceID types.ID, channelID, fromU if wh.headline == "" { return nil, http.StatusBadRequest, errors.Errorf("unsupported webhook") - } else if pluginConfig.DisplaySubscriptionNameInNotifications && subscriptionName != "" { - wh.headline = fmt.Sprintf("%s\nSubscription: **%s**", wh.headline, subscriptionName) + } + + // Keep the dedup identity independent of the subscription name so overlapping + // subscriptions on the same channel collapse into one post. + headline := wh.headline + if pluginConfig.DisplaySubscriptionNameInNotifications && subscriptionName != "" { + headline = fmt.Sprintf("%s\nSubscription: **%s**", headline, subscriptionName) } post := &model.Post{ @@ -119,14 +126,28 @@ func (wh webhook) PostToChannel(p *Plugin, instanceID types.ID, channelID, fromU { // TODO is this supposed to be themed? Color: "#95b7d0", - Fallback: wh.headline, - Pretext: wh.headline, + Fallback: headline, + Pretext: headline, Text: text, Fields: wh.fields, }, }) } else { - post.Message = wh.headline + post.Message = headline + } + + // Atomically claim the dedup key before posting so concurrent webhook + // deliveries can't both pass the check and post duplicates. SetAtomic(nil) + // only writes when the key does not already exist. + dedupKey := channelPostDedupKey(instanceID, &wh, channelID) + claimed, kvErr := p.client.KV.Set(dedupKey, true, pluginapi.SetExpiry(webhookDedupTTL), pluginapi.SetAtomic(nil)) + switch { + case kvErr != nil: + // Fail open: post rather than dropping the event. + p.client.Log.Warn("PostToChannel: failed to claim dedup key, posting anyway", "key", dedupKey, "error", kvErr.Error()) + case !claimed: + // Another delivery already claimed this post. + return nil, http.StatusOK, nil } if err := p.client.Post.CreatePost(post); err != nil { @@ -224,7 +245,7 @@ func (wh *webhook) PostNotifications(p *Plugin, instanceID types.ID) ([]*model.P // deliveries can't both pass the check and send duplicates. SetAtomic(nil) // only writes when the key does not already exist. dedupKey := notificationDedupKey(instance.GetID(), wh, mattermostUserID, notification.message) - claimed, kvErr := p.client.KV.Set(dedupKey, true, pluginapi.SetExpiry(notificationDedupTTL), pluginapi.SetAtomic(nil)) + claimed, kvErr := p.client.KV.Set(dedupKey, true, pluginapi.SetExpiry(webhookDedupTTL), pluginapi.SetAtomic(nil)) switch { case kvErr != nil: // Fail open: send the notification rather than dropping it. @@ -266,6 +287,17 @@ func notificationDedupKey(instanceID types.ID, wh *webhook, recipientID types.ID return fmt.Sprintf(notificationDedupKeyFmt, hex.EncodeToString(hash[:])) } +func channelPostDedupKey(instanceID types.ID, wh *webhook, channelID string) string { + var sb strings.Builder + fmt.Fprintf(&sb, "%s_%s_%s_%s_%s", + string(instanceID), wh.Issue.Key, channelID, wh.headline, wh.text) + for _, f := range wh.fields { + fmt.Fprintf(&sb, "_%s=%s", f.Title, f.Value) + } + hash := sha256.Sum256([]byte(sb.String())) + return fmt.Sprintf(channelPostDedupKeyFmt, hex.EncodeToString(hash[:])) +} + func (p *Plugin) GetWebhookURL(jiraURL string, teamID, channelID string) (subURL, legacyURL string, err error) { cf := p.getConfig() diff --git a/server/webhook_http_test.go b/server/webhook_http_test.go index 8f670fb4..0e6b6771 100644 --- a/server/webhook_http_test.go +++ b/server/webhook_http_test.go @@ -645,6 +645,7 @@ func TestWebhookHTTP(t *testing.T) { } else { api.On("KVGet", mock.AnythingOfType("string")).Return(nil, (*model.AppError)(nil)) } + api.On("KVSetWithOptions", mock.AnythingOfType("string"), mock.Anything, mock.Anything).Return(true, nil) api.On("LogDebug", mockAnythingOfTypeBatch("string", 11)...).Return(nil) api.On("LogWarn", mockAnythingOfTypeBatch("string", 10)...).Return(nil) diff --git a/server/webhook_parser_misc_test.go b/server/webhook_parser_misc_test.go index ecd23502..e793a75d 100644 --- a/server/webhook_parser_misc_test.go +++ b/server/webhook_parser_misc_test.go @@ -11,6 +11,7 @@ import ( "testing" jira "github.com/andygrunwald/go-jira" + "github.com/mattermost/mattermost/server/public/model" "github.com/stretchr/testify/assert" "github.com/stretchr/testify/require" @@ -337,3 +338,66 @@ func TestNotificationDedupKey(t *testing.T) { notificationDedupKey(instanceID, wh, "user-abc", "Actor **commented** on PROJ-1")) }) } + +func TestChannelPostDedupKey(t *testing.T) { + makeWebhook := func(issueKey, headline, text string, fields []*model.SlackAttachmentField) *webhook { + return &webhook{ + JiraWebhook: &JiraWebhook{ + Issue: jira.Issue{Key: issueKey}, + }, + headline: headline, + text: text, + fields: fields, + } + } + + const instanceID = types.ID("https://jira.example.com") + + t.Run("same instance, issue, channel and content produce the same key", func(t *testing.T) { + wh1 := makeWebhook("PROJ-1", "Actor **commented** on PROJ-1", "some comment", nil) + wh2 := makeWebhook("PROJ-1", "Actor **commented** on PROJ-1", "some comment", nil) + assert.Equal(t, + channelPostDedupKey(instanceID, wh1, "channel-abc"), + channelPostDedupKey(instanceID, wh2, "channel-abc")) + }) + + t.Run("different channels produce different keys", func(t *testing.T) { + wh := makeWebhook("PROJ-1", "Actor **commented** on PROJ-1", "some comment", nil) + assert.NotEqual(t, + channelPostDedupKey(instanceID, wh, "channel-abc"), + channelPostDedupKey(instanceID, wh, "channel-xyz")) + }) + + t.Run("different instances produce different keys", func(t *testing.T) { + wh := makeWebhook("PROJ-1", "Actor **commented** on PROJ-1", "some comment", nil) + assert.NotEqual(t, + channelPostDedupKey(types.ID("https://jira-a.example.com"), wh, "channel-abc"), + channelPostDedupKey(types.ID("https://jira-b.example.com"), wh, "channel-abc")) + }) + + t.Run("different issues produce different keys", func(t *testing.T) { + assert.NotEqual(t, + channelPostDedupKey(instanceID, makeWebhook("PROJ-1", "headline", "text", nil), "channel-abc"), + channelPostDedupKey(instanceID, makeWebhook("PROJ-2", "headline", "text", nil), "channel-abc")) + }) + + t.Run("different headlines produce different keys", func(t *testing.T) { + assert.NotEqual(t, + channelPostDedupKey(instanceID, makeWebhook("PROJ-1", "Actor **commented** on PROJ-1", "text", nil), "channel-abc"), + channelPostDedupKey(instanceID, makeWebhook("PROJ-1", "Actor **updated** PROJ-1", "text", nil), "channel-abc")) + }) + + t.Run("different text produce different keys", func(t *testing.T) { + assert.NotEqual(t, + channelPostDedupKey(instanceID, makeWebhook("PROJ-1", "headline", "first comment", nil), "channel-abc"), + channelPostDedupKey(instanceID, makeWebhook("PROJ-1", "headline", "second comment", nil), "channel-abc")) + }) + + t.Run("different fields produce different keys", func(t *testing.T) { + wh1 := makeWebhook("PROJ-1", "headline", "text", []*model.SlackAttachmentField{{Title: "Priority", Value: "High"}}) + wh2 := makeWebhook("PROJ-1", "headline", "text", []*model.SlackAttachmentField{{Title: "Priority", Value: "Low"}}) + assert.NotEqual(t, + channelPostDedupKey(instanceID, wh1, "channel-abc"), + channelPostDedupKey(instanceID, wh2, "channel-abc")) + }) +} diff --git a/server/webhook_test.go b/server/webhook_test.go new file mode 100644 index 00000000..358c3e32 --- /dev/null +++ b/server/webhook_test.go @@ -0,0 +1,106 @@ +// Copyright (c) 2017-present Mattermost, Inc. All Rights Reserved. +// See LICENSE.txt for license information. + +package main + +import ( + "sync" + "sync/atomic" + "testing" + + jira "github.com/andygrunwald/go-jira" + "github.com/mattermost/mattermost/server/public/model" + "github.com/mattermost/mattermost/server/public/plugin/plugintest" + "github.com/mattermost/mattermost/server/public/plugin/plugintest/mock" + "github.com/mattermost/mattermost/server/public/pluginapi" + "github.com/stretchr/testify/require" +) + +func newTestChannelWebhook() *webhook { + return &webhook{ + JiraWebhook: &JiraWebhook{ + Issue: jira.Issue{ID: "10001", Key: "PROJ-1"}, + }, + headline: "Actor **commented** on PROJ-1", + } +} + +func TestPostToChannelDeduplicatesConcurrentDeliveries(t *testing.T) { + t.Run("concurrent deliveries for the same channel post only once", func(t *testing.T) { + api := &plugintest.API{} + + var kvWinners atomic.Int32 + api.On("KVSetWithOptions", mock.AnythingOfType("string"), mock.Anything, mock.Anything).Return(func(string, []byte, model.PluginKVSetOptions) (bool, *model.AppError) { + if kvWinners.Add(1) == 1 { + return true, nil + } + return false, nil + }) + api.On("CreatePost", mock.AnythingOfType("*model.Post")).Return(&model.Post{Id: "post1"}, nil).Once() + + p := &Plugin{} + p.SetAPI(api) + p.client = pluginapi.NewClient(api, p.Driver) + + const callers = 25 + var wg sync.WaitGroup + wg.Add(callers) + start := make(chan struct{}) + for i := 0; i < callers; i++ { + go func() { + defer wg.Done() + <-start + _, _, err := newTestChannelWebhook().PostToChannel(p, "instance-1", "channel-1", "bot-user-id", "") + require.NoError(t, err) + }() + } + close(start) + wg.Wait() + + api.AssertExpectations(t) + }) + + t.Run("overlapping subscriptions on the same channel still post only once", func(t *testing.T) { + // Two subscriptions on the same channel produce identical webhook content + // but different subscription names; the dedup key must ignore the name. + api := &plugintest.API{} + var kvWinners atomic.Int32 + api.On("KVSetWithOptions", mock.AnythingOfType("string"), mock.Anything, mock.Anything).Return(func(string, []byte, model.PluginKVSetOptions) (bool, *model.AppError) { + if kvWinners.Add(1) == 1 { + return true, nil + } + return false, nil + }).Twice() + api.On("CreatePost", mock.AnythingOfType("*model.Post")).Return(&model.Post{Id: "post1"}, nil).Once() + + p := &Plugin{} + p.SetAPI(api) + p.client = pluginapi.NewClient(api, p.Driver) + + _, _, err := newTestChannelWebhook().PostToChannel(p, "instance-1", "channel-1", "bot-user-id", "subscription-a") + require.NoError(t, err) + + post, _, err := newTestChannelWebhook().PostToChannel(p, "instance-1", "channel-1", "bot-user-id", "subscription-b") + require.NoError(t, err) + require.Nil(t, post) + + api.AssertExpectations(t) + }) + + t.Run("different channels each get their own post", func(t *testing.T) { + api := &plugintest.API{} + api.On("KVSetWithOptions", mock.AnythingOfType("string"), mock.Anything, mock.Anything).Return(true, (*model.AppError)(nil)).Twice() + api.On("CreatePost", mock.AnythingOfType("*model.Post")).Return(&model.Post{Id: "post1"}, nil).Twice() + + p := &Plugin{} + p.SetAPI(api) + p.client = pluginapi.NewClient(api, p.Driver) + + _, _, err := newTestChannelWebhook().PostToChannel(p, "instance-1", "channel-1", "bot-user-id", "") + require.NoError(t, err) + _, _, err = newTestChannelWebhook().PostToChannel(p, "instance-1", "channel-2", "bot-user-id", "") + require.NoError(t, err) + + api.AssertExpectations(t) + }) +} From 86dfd6a33f1fa0b81ff543d3820e4f87f504900b Mon Sep 17 00:00:00 2001 From: avasconcelos114 Date: Fri, 14 Aug 2026 20:46:37 +0300 Subject: [PATCH 2/5] Adding debug logging to cases where notifications are skipped --- server/webhook.go | 1 + 1 file changed, 1 insertion(+) diff --git a/server/webhook.go b/server/webhook.go index a8ef65c8..5d2680e6 100644 --- a/server/webhook.go +++ b/server/webhook.go @@ -147,6 +147,7 @@ func (wh webhook) PostToChannel(p *Plugin, instanceID types.ID, channelID, fromU p.client.Log.Warn("PostToChannel: failed to claim dedup key, posting anyway", "key", dedupKey, "error", kvErr.Error()) case !claimed: // Another delivery already claimed this post. + p.client.Log.Debug("PostToChannel: another delivery already claimed this post, skipping", "key", dedupKey) return nil, http.StatusOK, nil } From 98f0fa6afe1a135ed642503823d2150cdb13ba21 Mon Sep 17 00:00:00 2001 From: avasconcelos114 Date: Fri, 14 Aug 2026 20:53:38 +0300 Subject: [PATCH 3/5] Added debug logs to test assertions --- server/webhook_test.go | 2 ++ 1 file changed, 2 insertions(+) diff --git a/server/webhook_test.go b/server/webhook_test.go index 358c3e32..2caf6a3e 100644 --- a/server/webhook_test.go +++ b/server/webhook_test.go @@ -37,6 +37,7 @@ func TestPostToChannelDeduplicatesConcurrentDeliveries(t *testing.T) { return false, nil }) api.On("CreatePost", mock.AnythingOfType("*model.Post")).Return(&model.Post{Id: "post1"}, nil).Once() + api.On("LogDebug", mockAnythingOfTypeBatch("string", 3)...).Return() p := &Plugin{} p.SetAPI(api) @@ -72,6 +73,7 @@ func TestPostToChannelDeduplicatesConcurrentDeliveries(t *testing.T) { return false, nil }).Twice() api.On("CreatePost", mock.AnythingOfType("*model.Post")).Return(&model.Post{Id: "post1"}, nil).Once() + api.On("LogDebug", mockAnythingOfTypeBatch("string", 3)...).Return().Once() p := &Plugin{} p.SetAPI(api) From 5221ff70694c0109a1497a2df69ed2942ca5edbf Mon Sep 17 00:00:00 2001 From: avasconcelos114 Date: Fri, 14 Aug 2026 21:11:55 +0300 Subject: [PATCH 4/5] Applying PR feedback --- server/webhook.go | 39 +++++++++++++++---------- server/webhook_parser_misc_test.go | 46 ++++++++++++++++++++++++++++++ server/webhook_test.go | 14 +++++++-- 3 files changed, 81 insertions(+), 18 deletions(-) diff --git a/server/webhook.go b/server/webhook.go index 5d2680e6..14591788 100644 --- a/server/webhook.go +++ b/server/webhook.go @@ -5,12 +5,13 @@ package main import ( "crypto/sha256" + "encoding/binary" "encoding/hex" "fmt" + "hash" "net/http" "net/url" "strconv" - "strings" "time" "github.com/pkg/errors" @@ -277,26 +278,34 @@ func newWebhook(jwh *JiraWebhook, eventType string, format string, args ...inter } } +// writeDedupToken writes s into h prefixed with its length, so concatenated +// tokens can't collide due to separator characters appearing inside values. +func writeDedupToken(h hash.Hash, s string) { + var length [8]byte + binary.BigEndian.PutUint64(length[:], uint64(len(s))) + h.Write(length[:]) //nolint:errcheck // hash.Hash.Write never returns an error + h.Write([]byte(s)) //nolint:errcheck // hash.Hash.Write never returns an error +} + func notificationDedupKey(instanceID types.ID, wh *webhook, recipientID types.ID, message string) string { - raw := fmt.Sprintf("%s_%s_%s_%s", - string(instanceID), - wh.Issue.Key, - string(recipientID), - message, - ) - hash := sha256.Sum256([]byte(raw)) - return fmt.Sprintf(notificationDedupKeyFmt, hex.EncodeToString(hash[:])) + h := sha256.New() + for _, token := range []string{string(instanceID), wh.Issue.Key, string(recipientID), message} { + writeDedupToken(h, token) + } + return fmt.Sprintf(notificationDedupKeyFmt, hex.EncodeToString(h.Sum(nil))) } func channelPostDedupKey(instanceID types.ID, wh *webhook, channelID string) string { - var sb strings.Builder - fmt.Fprintf(&sb, "%s_%s_%s_%s_%s", - string(instanceID), wh.Issue.Key, channelID, wh.headline, wh.text) + h := sha256.New() + for _, token := range []string{string(instanceID), wh.Issue.Key, channelID, wh.headline, wh.text} { + writeDedupToken(h, token) + } for _, f := range wh.fields { - fmt.Fprintf(&sb, "_%s=%s", f.Title, f.Value) + writeDedupToken(h, f.Title) + writeDedupToken(h, fmt.Sprintf("%v", f.Value)) + writeDedupToken(h, fmt.Sprintf("%t", bool(f.Short))) } - hash := sha256.Sum256([]byte(sb.String())) - return fmt.Sprintf(channelPostDedupKeyFmt, hex.EncodeToString(hash[:])) + return fmt.Sprintf(channelPostDedupKeyFmt, hex.EncodeToString(h.Sum(nil))) } func (p *Plugin) GetWebhookURL(jiraURL string, teamID, channelID string) (subURL, legacyURL string, err error) { diff --git a/server/webhook_parser_misc_test.go b/server/webhook_parser_misc_test.go index e793a75d..89d19111 100644 --- a/server/webhook_parser_misc_test.go +++ b/server/webhook_parser_misc_test.go @@ -337,6 +337,21 @@ func TestNotificationDedupKey(t *testing.T) { notificationDedupKey(instanceID, wh, "user-abc", "Actor **assigned** you to PROJ-1"), notificationDedupKey(instanceID, wh, "user-abc", "Actor **commented** on PROJ-1")) }) + + t.Run("values containing separator-like characters do not collide", func(t *testing.T) { + // A naive "_"-joined encoding would make ("user-abc", "extra_hello") + // collide with ("user-abc_extra", "hello"). The length-prefixed encoding + // must keep these distinct. + wh := makeWebhook("PROJ-1") + assert.NotEqual(t, + notificationDedupKey(instanceID, wh, "user-abc", "extra_hello"), + notificationDedupKey(instanceID, wh, "user-abc_extra", "hello")) + + // Same idea across the issue key/recipient boundary. + assert.NotEqual(t, + notificationDedupKey(instanceID, makeWebhook("PROJ-1_user"), "abc", "hello"), + notificationDedupKey(instanceID, makeWebhook("PROJ-1"), "user_abc", "hello")) + }) } func TestChannelPostDedupKey(t *testing.T) { @@ -400,4 +415,35 @@ func TestChannelPostDedupKey(t *testing.T) { channelPostDedupKey(instanceID, wh1, "channel-abc"), channelPostDedupKey(instanceID, wh2, "channel-abc")) }) + + t.Run("different field Short flags produce different keys", func(t *testing.T) { + wh1 := makeWebhook("PROJ-1", "headline", "text", []*model.SlackAttachmentField{{Title: "Priority", Value: "High", Short: true}}) + wh2 := makeWebhook("PROJ-1", "headline", "text", []*model.SlackAttachmentField{{Title: "Priority", Value: "High", Short: false}}) + assert.NotEqual(t, + channelPostDedupKey(instanceID, wh1, "channel-abc"), + channelPostDedupKey(instanceID, wh2, "channel-abc")) + }) + + t.Run("values containing separator-like characters do not collide", func(t *testing.T) { + // A naive "_"-joined encoding would make ("A_B", "C") collide with ("A", "B_C"). + // The length-prefixed encoding must keep these distinct. + wh1 := makeWebhook("PROJ-1", "A_B", "C", nil) + wh2 := makeWebhook("PROJ-1", "A", "B_C", nil) + assert.NotEqual(t, + channelPostDedupKey(instanceID, wh1, "channel-abc"), + channelPostDedupKey(instanceID, wh2, "channel-abc")) + + // Same idea across the channelID/headline boundary. + wh3 := makeWebhook("PROJ-1", "B", "text", nil) + assert.NotEqual(t, + channelPostDedupKey(instanceID, wh3, "channel-abc_extra"), + channelPostDedupKey(instanceID, makeWebhook("PROJ-1", "extra_B", "text", nil), "channel-abc")) + + // And across a field's title/value boundary. + wh4 := makeWebhook("PROJ-1", "headline", "text", []*model.SlackAttachmentField{{Title: "A=B", Value: "C"}}) + wh5 := makeWebhook("PROJ-1", "headline", "text", []*model.SlackAttachmentField{{Title: "A", Value: "B=C"}}) + assert.NotEqual(t, + channelPostDedupKey(instanceID, wh4, "channel-abc"), + channelPostDedupKey(instanceID, wh5, "channel-abc")) + }) } diff --git a/server/webhook_test.go b/server/webhook_test.go index 2caf6a3e..852b94e8 100644 --- a/server/webhook_test.go +++ b/server/webhook_test.go @@ -7,6 +7,7 @@ import ( "sync" "sync/atomic" "testing" + "time" jira "github.com/andygrunwald/go-jira" "github.com/mattermost/mattermost/server/public/model" @@ -25,12 +26,19 @@ func newTestChannelWebhook() *webhook { } } +// isDedupClaimOptions matches the KV options PostToChannel's dedup claim must +// use: an atomic write (no old value expected, since the key shouldn't yet +// exist) with the shared webhook dedup TTL. +func isDedupClaimOptions(opts model.PluginKVSetOptions) bool { + return opts.Atomic && opts.OldValue == nil && opts.ExpireInSeconds == int64(webhookDedupTTL/time.Second) +} + func TestPostToChannelDeduplicatesConcurrentDeliveries(t *testing.T) { t.Run("concurrent deliveries for the same channel post only once", func(t *testing.T) { api := &plugintest.API{} var kvWinners atomic.Int32 - api.On("KVSetWithOptions", mock.AnythingOfType("string"), mock.Anything, mock.Anything).Return(func(string, []byte, model.PluginKVSetOptions) (bool, *model.AppError) { + api.On("KVSetWithOptions", mock.AnythingOfType("string"), mock.Anything, mock.MatchedBy(isDedupClaimOptions)).Return(func(string, []byte, model.PluginKVSetOptions) (bool, *model.AppError) { if kvWinners.Add(1) == 1 { return true, nil } @@ -66,7 +74,7 @@ func TestPostToChannelDeduplicatesConcurrentDeliveries(t *testing.T) { // but different subscription names; the dedup key must ignore the name. api := &plugintest.API{} var kvWinners atomic.Int32 - api.On("KVSetWithOptions", mock.AnythingOfType("string"), mock.Anything, mock.Anything).Return(func(string, []byte, model.PluginKVSetOptions) (bool, *model.AppError) { + api.On("KVSetWithOptions", mock.AnythingOfType("string"), mock.Anything, mock.MatchedBy(isDedupClaimOptions)).Return(func(string, []byte, model.PluginKVSetOptions) (bool, *model.AppError) { if kvWinners.Add(1) == 1 { return true, nil } @@ -91,7 +99,7 @@ func TestPostToChannelDeduplicatesConcurrentDeliveries(t *testing.T) { t.Run("different channels each get their own post", func(t *testing.T) { api := &plugintest.API{} - api.On("KVSetWithOptions", mock.AnythingOfType("string"), mock.Anything, mock.Anything).Return(true, (*model.AppError)(nil)).Twice() + api.On("KVSetWithOptions", mock.AnythingOfType("string"), mock.Anything, mock.MatchedBy(isDedupClaimOptions)).Return(true, (*model.AppError)(nil)).Twice() api.On("CreatePost", mock.AnythingOfType("*model.Post")).Return(&model.Post{Id: "post1"}, nil).Twice() p := &Plugin{} From 1b5583a57457e254efb75eb42127682914056b14 Mon Sep 17 00:00:00 2001 From: avasconcelos114 Date: Mon, 17 Aug 2026 15:13:27 +0300 Subject: [PATCH 5/5] Applying PR Feedback --- server/webhook.go | 19 +++++-- server/webhook_test.go | 116 ++++++++++++++++++++++++++++++----------- 2 files changed, 103 insertions(+), 32 deletions(-) diff --git a/server/webhook.go b/server/webhook.go index 14591788..7eccc969 100644 --- a/server/webhook.go +++ b/server/webhook.go @@ -153,6 +153,9 @@ func (wh webhook) PostToChannel(p *Plugin, instanceID types.ID, channelID, fromU } if err := p.client.Post.CreatePost(post); err != nil { + if claimed { + p.releaseDedupKey("PostToChannel", dedupKey) + } return nil, http.StatusInternalServerError, err } @@ -260,9 +263,9 @@ func (wh *webhook) PostNotifications(p *Plugin, instanceID types.ID) ([]*model.P post, err := p.CreateBotDMPost(instance.GetID(), mattermostUserID, notification.message, notification.postType) if err != nil { p.errorf("PostNotifications: failed to create notification post, err: %v", err) - // 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. + if claimed { + p.releaseDedupKey("PostNotifications", dedupKey) + } continue } posts = append(posts, post) @@ -278,6 +281,16 @@ func newWebhook(jwh *JiraWebhook, eventType string, format string, args ...inter } } +// releaseDedupKey drops a dedup claim after the post it guarded failed, so a +// later delivery of the same event can retry instead of being skipped until the +// claim expires. Only safe to call when this delivery won the claim. +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 after a failed post; duplicates of this event will be skipped until the claim expires", + "key", dedupKey, "error", err.Error()) + } +} + // writeDedupToken writes s into h prefixed with its length, so concatenated // tokens can't collide due to separator characters appearing inside values. func writeDedupToken(h hash.Hash, s string) { diff --git a/server/webhook_test.go b/server/webhook_test.go index 852b94e8..7cc00a0a 100644 --- a/server/webhook_test.go +++ b/server/webhook_test.go @@ -5,7 +5,6 @@ package main import ( "sync" - "sync/atomic" "testing" "time" @@ -26,30 +25,71 @@ func newTestChannelWebhook() *webhook { } } -// isDedupClaimOptions matches the KV options PostToChannel's dedup claim must -// use: an atomic write (no old value expected, since the key shouldn't yet -// exist) with the shared webhook dedup TTL. +// isDedupClaimOptions matches the KV options a dedup claim must use: an atomic +// write (no old value expected, since the key shouldn't yet exist) with the +// shared webhook dedup TTL. func isDedupClaimOptions(opts model.PluginKVSetOptions) bool { return opts.Atomic && opts.OldValue == nil && opts.ExpireInSeconds == int64(webhookDedupTTL/time.Second) } -func TestPostToChannelDeduplicatesConcurrentDeliveries(t *testing.T) { - t.Run("concurrent deliveries for the same channel post only once", func(t *testing.T) { - api := &plugintest.API{} +// isDedupReleaseOptions matches the KV options produced by KV.Delete, which is +// implemented as a plain non-atomic write of a nil value. +func isDedupReleaseOptions(opts model.PluginKVSetOptions) bool { + return !opts.Atomic && opts.ExpireInSeconds == 0 +} - var kvWinners atomic.Int32 - api.On("KVSetWithOptions", mock.AnythingOfType("string"), mock.Anything, mock.MatchedBy(isDedupClaimOptions)).Return(func(string, []byte, model.PluginKVSetOptions) (bool, *model.AppError) { - if kvWinners.Add(1) == 1 { - return true, nil +// fakeDedupKV emulates the KV store's atomic-claim semantics keyed on the actual +// dedup key, so tests assert on the dedup identity itself rather than on call +// counts. 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 } - return false, nil + claimed[key] = true + return true, nil }) + + api.On("KVSetWithOptions", mock.AnythingOfType("string"), mock.Anything, 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 +} + +func TestPostToChannelDeduplicatesConcurrentDeliveries(t *testing.T) { + t.Run("concurrent deliveries for the same channel post only once", func(t *testing.T) { + api := &plugintest.API{} + claimedKeys := fakeDedupKV(api) api.On("CreatePost", mock.AnythingOfType("*model.Post")).Return(&model.Post{Id: "post1"}, nil).Once() api.On("LogDebug", mockAnythingOfTypeBatch("string", 3)...).Return() - p := &Plugin{} - p.SetAPI(api) - p.client = pluginapi.NewClient(api, p.Driver) + p := newTestPluginWithAPI(api) const callers = 25 var wg sync.WaitGroup @@ -66,6 +106,8 @@ func TestPostToChannelDeduplicatesConcurrentDeliveries(t *testing.T) { close(start) wg.Wait() + // One key means every delivery computed the same dedup identity. + require.Len(t, claimedKeys(), 1) api.AssertExpectations(t) }) @@ -73,19 +115,11 @@ func TestPostToChannelDeduplicatesConcurrentDeliveries(t *testing.T) { // Two subscriptions on the same channel produce identical webhook content // but different subscription names; the dedup key must ignore the name. api := &plugintest.API{} - var kvWinners atomic.Int32 - api.On("KVSetWithOptions", mock.AnythingOfType("string"), mock.Anything, mock.MatchedBy(isDedupClaimOptions)).Return(func(string, []byte, model.PluginKVSetOptions) (bool, *model.AppError) { - if kvWinners.Add(1) == 1 { - return true, nil - } - return false, nil - }).Twice() + claimedKeys := fakeDedupKV(api) api.On("CreatePost", mock.AnythingOfType("*model.Post")).Return(&model.Post{Id: "post1"}, nil).Once() api.On("LogDebug", mockAnythingOfTypeBatch("string", 3)...).Return().Once() - p := &Plugin{} - p.SetAPI(api) - p.client = pluginapi.NewClient(api, p.Driver) + p := newTestPluginWithAPI(api) _, _, err := newTestChannelWebhook().PostToChannel(p, "instance-1", "channel-1", "bot-user-id", "subscription-a") require.NoError(t, err) @@ -94,23 +128,47 @@ func TestPostToChannelDeduplicatesConcurrentDeliveries(t *testing.T) { require.NoError(t, err) require.Nil(t, post) + require.Len(t, claimedKeys(), 1) api.AssertExpectations(t) }) t.Run("different channels each get their own post", func(t *testing.T) { api := &plugintest.API{} - api.On("KVSetWithOptions", mock.AnythingOfType("string"), mock.Anything, mock.MatchedBy(isDedupClaimOptions)).Return(true, (*model.AppError)(nil)).Twice() + claimedKeys := fakeDedupKV(api) api.On("CreatePost", mock.AnythingOfType("*model.Post")).Return(&model.Post{Id: "post1"}, nil).Twice() - p := &Plugin{} - p.SetAPI(api) - p.client = pluginapi.NewClient(api, p.Driver) + p := newTestPluginWithAPI(api) _, _, err := newTestChannelWebhook().PostToChannel(p, "instance-1", "channel-1", "bot-user-id", "") require.NoError(t, err) _, _, err = newTestChannelWebhook().PostToChannel(p, "instance-1", "channel-2", "bot-user-id", "") require.NoError(t, err) + // Two distinct keys means the channel ID is part of the dedup identity. + require.Len(t, claimedKeys(), 2) + api.AssertExpectations(t) + }) + + t.Run("a failed post releases the claim so the event can be redelivered", func(t *testing.T) { + api := &plugintest.API{} + claimedKeys := fakeDedupKV(api) + api.On("CreatePost", mock.AnythingOfType("*model.Post")). + Return((*model.Post)(nil), model.NewAppError("CreatePost", "boom", nil, "", 500)).Once() + + p := newTestPluginWithAPI(api) + + _, status, err := newTestChannelWebhook().PostToChannel(p, "instance-1", "channel-1", "bot-user-id", "") + require.Error(t, err) + require.Equal(t, 500, status) + require.Empty(t, claimedKeys(), "the claim should be released when the post fails") + + // A redelivery of the same event now succeeds instead of being skipped. + api.On("CreatePost", mock.AnythingOfType("*model.Post")).Return(&model.Post{Id: "post1"}, nil).Once() + + post, _, err := newTestChannelWebhook().PostToChannel(p, "instance-1", "channel-1", "bot-user-id", "") + require.NoError(t, err) + require.NotNil(t, post) + require.Len(t, claimedKeys(), 1) api.AssertExpectations(t) }) }