Function TestBackpressure

coderd/notifications/notifications_test.go:269–391

TestBackpressure validates that delays in processing the buffered updates will result in slowed dequeue rates. As a side-effect, this also tests the graceful shutdown and flushing of the buffers.

(t *testing.T)

// TestBackpressure validates that delays in processing the buffered updates will result in slowed dequeue rates.
// As a side-effect, this also tests the graceful shutdown and flushing of the buffers.
func TestBackpressure(t *testing.T) {
	t.Parallel()

	store, pubsub := dbtestutil.NewDB(t)
	logger := testutil.Logger(t)
	ctx := dbauthz.AsNotifier(testutil.Context(t, testutil.WaitShort))

	const method = database.NotificationMethodWebhook
	cfg := defaultNotificationsConfig(method)

	// Tune the queue to fetch often.
	const fetchInterval = time.Millisecond * 200
	const batchSize = 10
	cfg.FetchInterval = serpent.Duration(fetchInterval)
	cfg.LeaseCount = serpent.Int64(batchSize)
	// never time out for this test
	cfg.LeasePeriod = serpent.Duration(time.Hour)
	cfg.DispatchTimeout = serpent.Duration(time.Hour - time.Millisecond)

	// Shrink buffers down and increase flush interval to provoke backpressure.
	// Flush buffers every 5 fetch intervals.
	const syncInterval = time.Second
	cfg.StoreSyncInterval = serpent.Duration(syncInterval)
	cfg.StoreSyncBufferSize = serpent.Int64(2)

	handler := &chanHandler{calls: make(chan dispatchCall)}

	// Intercept calls to submit the buffered updates to the store.
	storeInterceptor := &syncInterceptor{Store: store}

	mClock := quartz.NewMock(t)
	syncTrap := mClock.Trap().NewTicker("Manager", "storeSync")
	defer syncTrap.Close()
	fetchTrap := mClock.Trap().TickerFunc("notifier", "fetchInterval")
	defer fetchTrap.Close()

	// GIVEN: a notification manager whose updates will be intercepted
	mgr, err := notifications.NewManager(cfg, storeInterceptor, pubsub, defaultHelpers(), createMetrics(),
		logger.Named("manager"), notifications.WithTestClock(mClock))
	require.NoError(t, err)
	mgr.WithHandlers(map[database.NotificationMethod]notifications.Handler{
		method:                           handler,
		database.NotificationMethodInbox: handler,
	})
	enq, err := notifications.NewStoreEnqueuer(cfg, store, defaultHelpers(), logger.Named("enqueuer"), mClock)
	require.NoError(t, err)

	user := createSampleUser(t, store)

	// WHEN: a set of notifications are enqueued, which causes backpressure due to the batchSize which can be processed per fetch
	const totalMessages = 30
	for i := range totalMessages {
		_, err = enq.Enqueue(ctx, user.ID, notifications.TemplateWorkspaceDeleted, map[string]string{"i": fmt.Sprintf("%d", i)}, "test")
		require.NoError(t, err)
	}

	// Start the notifier.
	mgr.Run(ctx)