opensource/coder
3
0
Fork 0
mirror of https://github.com/coder/coder.git synced 2026-09-24 15:04:27 +08:00
Code Issues Packages Projects Releases Wiki Activity

feat: add notification deduplication trigger (#14172)

Browse Source
This commit is contained in:
Danny Kopping
2024-08-21 11:18:03 +02:00
committed by GitHub
parent d9f419308a
commit 9c8c6a952d
12 changed files with 168 additions and 28 deletions
Download Patch File Download Diff File Expand all files Collapse all files
coderd/notifications/enqueuer.go
+18 -2
View File
@@ -10,14 +10,19 @@ import (
"golang.org/x/xerrors"
"cdr.dev/slog"
"github.com/coder/quartz"
"github.com/coder/coder/v2/coderd/database"
"github.com/coder/coder/v2/coderd/database/dbtime"
"github.com/coder/coder/v2/coderd/notifications/render"
"github.com/coder/coder/v2/coderd/notifications/types"
"github.com/coder/coder/v2/codersdk"
)
var ErrCannotEnqueueDisabledNotification = xerrors.New("user has disabled this notification")
var (
ErrCannotEnqueueDisabledNotification = xerrors.New("user has disabled this notification")
ErrDuplicate = xerrors.New("duplicate notification")
)
type StoreEnqueuer struct {
store Store
@@ -27,10 +32,12 @@ type StoreEnqueuer struct {
// helpers holds a map of template funcs which are used when rendering templates. These need to be passed in because
// the template funcs will return values which are inappropriately encapsulated in this struct.
helpers template.FuncMap
// Used to manipulate time in tests.
clock quartz.Clock
}
// NewStoreEnqueuer creates an Enqueuer implementation which can persist notification messages in the store.
func NewStoreEnqueuer(cfg codersdk.NotificationsConfig, store Store, helpers template.FuncMap, log slog.Logger) (*StoreEnqueuer, error) {
func NewStoreEnqueuer(cfg codersdk.NotificationsConfig, store Store, helpers template.FuncMap, log slog.Logger, clock quartz.Clock) (*StoreEnqueuer, error) {
var method database.NotificationMethod
if err := method.Scan(cfg.Method.String()); err != nil {
return nil, xerrors.Errorf("given notification method %q is invalid", cfg.Method)
@@ -41,6 +48,7 @@ func NewStoreEnqueuer(cfg codersdk.NotificationsConfig, store Store, helpers tem
log: log,
defaultMethod: method,
helpers: helpers,
clock: clock,
}, nil
}
@@ -81,6 +89,7 @@ func (s *StoreEnqueuer) Enqueue(ctx context.Context, userID, templateID uuid.UUI
Payload: input,
Targets: targets,
CreatedBy: createdBy,
CreatedAt: dbtime.Time(s.clock.Now().UTC()),
})
if err != nil {
// We have a trigger on the notification_messages table named `inhibit_enqueue_if_disabled` which prevents messages
@@ -92,6 +101,13 @@ func (s *StoreEnqueuer) Enqueue(ctx context.Context, userID, templateID uuid.UUI
return nil, ErrCannotEnqueueDisabledNotification
}
// If the enqueue fails due to a dedupe hash conflict, this means that a notification has already been enqueued
// today with identical properties. It's far simpler to prevent duplicate sends in this central manner, rather than
// having each notification enqueue handle its own logic.
if database.IsUniqueViolation(err, database.UniqueNotificationMessagesDedupeHashIndex) {
return nil, ErrDuplicate
}
s.log.Warn(ctx, "failed to enqueue notification", slog.F("template_id", templateID), slog.F("input", input), slog.Error(err))
return nil, xerrors.Errorf("enqueue notification: %w", err)
}
coderd/notifications/manager_test.go
+5 -3
View File
@@ -12,13 +12,15 @@ import (
"github.com/stretchr/testify/require"
"golang.org/x/xerrors"
"github.com/coder/quartz"
"github.com/coder/serpent"
"github.com/coder/coder/v2/coderd/database"
"github.com/coder/coder/v2/coderd/database/dbgen"
"github.com/coder/coder/v2/coderd/notifications"
"github.com/coder/coder/v2/coderd/notifications/dispatch"
"github.com/coder/coder/v2/coderd/notifications/types"
"github.com/coder/coder/v2/testutil"
"github.com/coder/serpent"
)
func TestBufferedUpdates(t *testing.T) {
@@ -39,7 +41,7 @@ func TestBufferedUpdates(t *testing.T) {
mgr.WithHandlers(map[database.NotificationMethod]notifications.Handler{
database.NotificationMethodSmtp: santa,
})
enq, err := notifications.NewStoreEnqueuer(cfg, interceptor, defaultHelpers(), logger.Named("notifications-enqueuer"))
enq, err := notifications.NewStoreEnqueuer(cfg, interceptor, defaultHelpers(), logger.Named("notifications-enqueuer"), quartz.NewReal())
require.NoError(t, err)
user := dbgen.User(t, db, database.User{})
@@ -127,7 +129,7 @@ func TestBuildPayload(t *testing.T) {
}
})
enq, err := notifications.NewStoreEnqueuer(defaultNotificationsConfig(database.NotificationMethodSmtp), interceptor, helpers, logger.Named("notifications-enqueuer"))
enq, err := notifications.NewStoreEnqueuer(defaultNotificationsConfig(database.NotificationMethodSmtp), interceptor, helpers, logger.Named("notifications-enqueuer"), quartz.NewReal())
require.NoError(t, err)
// WHEN: a notification is enqueued
coderd/notifications/metrics_test.go
+6 -4
View File
@@ -13,6 +13,8 @@ import (
"github.com/stretchr/testify/assert"
"github.com/stretchr/testify/require"
"github.com/coder/quartz"
"github.com/coder/serpent"
"github.com/coder/coder/v2/coderd/database"
@@ -61,7 +63,7 @@ func TestMetrics(t *testing.T) {
method: handler,
})
enq, err := notifications.NewStoreEnqueuer(cfg, store, defaultHelpers(), logger.Named("enqueuer"))
enq, err := notifications.NewStoreEnqueuer(cfg, store, defaultHelpers(), logger.Named("enqueuer"), quartz.NewReal())
require.NoError(t, err)
user := createSampleUser(t, store)
@@ -228,7 +230,7 @@ func TestPendingUpdatesMetric(t *testing.T) {
method: handler,
})
enq, err := notifications.NewStoreEnqueuer(cfg, store, defaultHelpers(), logger.Named("enqueuer"))
enq, err := notifications.NewStoreEnqueuer(cfg, store, defaultHelpers(), logger.Named("enqueuer"), quartz.NewReal())
require.NoError(t, err)
user := createSampleUser(t, store)
@@ -305,7 +307,7 @@ func TestInflightDispatchesMetric(t *testing.T) {
method: delayer,
})
enq, err := notifications.NewStoreEnqueuer(cfg, store, defaultHelpers(), logger.Named("enqueuer"))
enq, err := notifications.NewStoreEnqueuer(cfg, store, defaultHelpers(), logger.Named("enqueuer"), quartz.NewReal())
require.NoError(t, err)
user := createSampleUser(t, store)
@@ -384,7 +386,7 @@ func TestCustomMethodMetricCollection(t *testing.T) {
customMethod: webhookHandler,
})
enq, err := notifications.NewStoreEnqueuer(cfg, store, defaultHelpers(), logger.Named("enqueuer"))
enq, err := notifications.NewStoreEnqueuer(cfg, store, defaultHelpers(), logger.Named("enqueuer"), quartz.NewReal())
require.NoError(t, err)
user := createSampleUser(t, store)
coderd/notifications/notifications_test.go
+61 -11
View File
@@ -21,6 +21,8 @@ import (
"golang.org/x/xerrors"
"github.com/coder/quartz"
"github.com/google/uuid"
smtpmock "github.com/mocktools/go-smtp-mock/v2"
"github.com/stretchr/testify/assert"
@@ -71,7 +73,7 @@ func TestBasicNotificationRoundtrip(t *testing.T) {
t.Cleanup(func() {
assert.NoError(t, mgr.Stop(ctx))
})
enq, err := notifications.NewStoreEnqueuer(cfg, db, defaultHelpers(), logger.Named("enqueuer"))
enq, err := notifications.NewStoreEnqueuer(cfg, db, defaultHelpers(), logger.Named("enqueuer"), quartz.NewReal())
require.NoError(t, err)
user := createSampleUser(t, db)
@@ -145,7 +147,7 @@ func TestSMTPDispatch(t *testing.T) {
t.Cleanup(func() {
assert.NoError(t, mgr.Stop(ctx))
})
enq, err := notifications.NewStoreEnqueuer(cfg, db, defaultHelpers(), logger.Named("enqueuer"))
enq, err := notifications.NewStoreEnqueuer(cfg, db, defaultHelpers(), logger.Named("enqueuer"), quartz.NewReal())
require.NoError(t, err)
user := createSampleUser(t, db)
@@ -205,7 +207,7 @@ func TestWebhookDispatch(t *testing.T) {
t.Cleanup(func() {
assert.NoError(t, mgr.Stop(ctx))
})
enq, err := notifications.NewStoreEnqueuer(cfg, db, defaultHelpers(), logger.Named("enqueuer"))
enq, err := notifications.NewStoreEnqueuer(cfg, db, defaultHelpers(), logger.Named("enqueuer"), quartz.NewReal())
require.NoError(t, err)
const (
@@ -301,7 +303,7 @@ func TestBackpressure(t *testing.T) {
mgr, err := notifications.NewManager(cfg, storeInterceptor, defaultHelpers(), createMetrics(), logger.Named("manager"))
require.NoError(t, err)
mgr.WithHandlers(map[database.NotificationMethod]notifications.Handler{method: handler})
enq, err := notifications.NewStoreEnqueuer(cfg, db, defaultHelpers(), logger.Named("enqueuer"))
enq, err := notifications.NewStoreEnqueuer(cfg, db, defaultHelpers(), logger.Named("enqueuer"), quartz.NewReal())
require.NoError(t, err)
user := createSampleUser(t, db)
@@ -399,7 +401,7 @@ func TestRetries(t *testing.T) {
assert.NoError(t, mgr.Stop(ctx))
})
mgr.WithHandlers(map[database.NotificationMethod]notifications.Handler{method: handler})
enq, err := notifications.NewStoreEnqueuer(cfg, db, defaultHelpers(), logger.Named("enqueuer"))
enq, err := notifications.NewStoreEnqueuer(cfg, db, defaultHelpers(), logger.Named("enqueuer"), quartz.NewReal())
require.NoError(t, err)
user := createSampleUser(t, db)
@@ -456,7 +458,7 @@ func TestExpiredLeaseIsRequeued(t *testing.T) {
mgr, err := notifications.NewManager(cfg, noopInterceptor, defaultHelpers(), createMetrics(), logger.Named("manager"))
require.NoError(t, err)
enq, err := notifications.NewStoreEnqueuer(cfg, db, defaultHelpers(), logger.Named("enqueuer"))
enq, err := notifications.NewStoreEnqueuer(cfg, db, defaultHelpers(), logger.Named("enqueuer"), quartz.NewReal())
require.NoError(t, err)
user := createSampleUser(t, db)
@@ -464,7 +466,8 @@ func TestExpiredLeaseIsRequeued(t *testing.T) {
// WHEN: a few notifications are enqueued which will all succeed
var msgs []string
for i := 0; i < msgCount; i++ {
id, err := enq.Enqueue(ctx, user.ID, notifications.TemplateWorkspaceDeleted, map[string]string{"type": "success"}, "test")
id, err := enq.Enqueue(ctx, user.ID, notifications.TemplateWorkspaceDeleted,
map[string]string{"type": "success", "index": fmt.Sprintf("%d", i)}, "test")
require.NoError(t, err)
msgs = append(msgs, id.String())
}
@@ -566,7 +569,7 @@ func TestNotifierPaused(t *testing.T) {
t.Cleanup(func() {
assert.NoError(t, mgr.Stop(ctx))
})
enq, err := notifications.NewStoreEnqueuer(cfg, db, defaultHelpers(), logger.Named("enqueuer"))
enq, err := notifications.NewStoreEnqueuer(cfg, db, defaultHelpers(), logger.Named("enqueuer"), quartz.NewReal())
require.NoError(t, err)
mgr.Run(ctx)
@@ -821,7 +824,7 @@ func TestDisabledBeforeEnqueue(t *testing.T) {
// GIVEN: an enqueuer & a sample user
cfg := defaultNotificationsConfig(database.NotificationMethodSmtp)
enq, err := notifications.NewStoreEnqueuer(cfg, db, defaultHelpers(), logger.Named("enqueuer"))
enq, err := notifications.NewStoreEnqueuer(cfg, db, defaultHelpers(), logger.Named("enqueuer"), quartz.NewReal())
require.NoError(t, err)
user := createSampleUser(t, db)
@@ -861,7 +864,7 @@ func TestDisabledAfterEnqueue(t *testing.T) {
assert.NoError(t, mgr.Stop(ctx))
})
enq, err := notifications.NewStoreEnqueuer(cfg, db, defaultHelpers(), logger.Named("enqueuer"))
enq, err := notifications.NewStoreEnqueuer(cfg, db, defaultHelpers(), logger.Named("enqueuer"), quartz.NewReal())
require.NoError(t, err)
user := createSampleUser(t, db)
@@ -967,7 +970,7 @@ func TestCustomNotificationMethod(t *testing.T) {
_ = mgr.Stop(ctx)
})
enq, err := notifications.NewStoreEnqueuer(cfg, db, defaultHelpers(), logger)
enq, err := notifications.NewStoreEnqueuer(cfg, db, defaultHelpers(), logger, quartz.NewReal())
require.NoError(t, err)
// WHEN: a notification of that template is enqueued, it should be delivered with the configured method - not the default.
@@ -1033,6 +1036,53 @@ func createOpts(t *testing.T) *coderdtest.Options {
}
}
// TestNotificationDuplicates validates that identical notifications cannot be sent on the same day.
func TestNotificationDuplicates(t *testing.T) {
t.Parallel()
// SETUP
if !dbtestutil.WillUsePostgres() {
t.Skip("This test requires postgres; it is testing the dedupe hash trigger in the database")
}
ctx, logger, db := setup(t)
method := database.NotificationMethodSmtp
cfg := defaultNotificationsConfig(method)
mgr, err := notifications.NewManager(cfg, db, defaultHelpers(), createMetrics(), logger.Named("manager"))
require.NoError(t, err)
t.Cleanup(func() {
assert.NoError(t, mgr.Stop(ctx))
})
// Set the time to a known value.
mClock := quartz.NewMock(t)
mClock.Set(time.Date(2024, 1, 15, 9, 0, 0, 0, time.UTC))
enq, err := notifications.NewStoreEnqueuer(cfg, db, defaultHelpers(), logger.Named("enqueuer"), mClock)
require.NoError(t, err)
user := createSampleUser(t, db)
// GIVEN: two notifications are enqueued with identical properties.
_, err = enq.Enqueue(ctx, user.ID, notifications.TemplateWorkspaceDeleted,
map[string]string{"initiator": "danny"}, "test", user.ID)
require.NoError(t, err)
// WHEN: the second is enqueued, the enqueuer will reject the request.
_, err = enq.Enqueue(ctx, user.ID, notifications.TemplateWorkspaceDeleted,
map[string]string{"initiator": "danny"}, "test", user.ID)
require.ErrorIs(t, err, notifications.ErrDuplicate)
// THEN: when the clock is advanced 24h, the notification will be accepted.
// NOTE: the time is used in the dedupe hash, so by advancing 24h we're creating a distinct notification from the one
// which was enqueued "yesterday".
mClock.Advance(time.Hour * 24)
_, err = enq.Enqueue(ctx, user.ID, notifications.TemplateWorkspaceDeleted,
map[string]string{"initiator": "danny"}, "test", user.ID)
require.NoError(t, err)
}
type fakeHandler struct {
mu sync.RWMutex
succeeded, failed []string