Function TestExpiredLeaseIsRequeued

coderd/notifications/notifications_test.go:488–588

TestExpiredLeaseIsRequeued validates that notification messages which are left in "leased" status will be requeued once their lease expires. "leased" is the status which messages are set to when they are acquired for processing, and this should not be a terminal state unless the Manager shuts down u

(t *testing.T)

// state unless the Manager shuts down ungracefully; the Manager is responsible for updating these messages' statuses once
// they have been processed.
func TestExpiredLeaseIsRequeued(t *testing.T) {
	t.Parallel()

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

	// GIVEN: a manager which has its updates intercepted and paused until measurements can be taken

	const (
		leasePeriod = time.Second
		msgCount    = 5
		method      = database.NotificationMethodSmtp
	)

	cfg := defaultNotificationsConfig(method)
	// Set low lease period to speed up tests.
	cfg.LeasePeriod = serpent.Duration(leasePeriod)
	cfg.DispatchTimeout = serpent.Duration(leasePeriod - time.Millisecond)

	noopInterceptor := newNoopStoreSyncer(store)

	mgrCtx, cancelManagerCtx := context.WithCancel(dbauthz.AsNotifier(context.Background()))
	t.Cleanup(cancelManagerCtx)

	mgr, err := notifications.NewManager(cfg, noopInterceptor, pubsub, defaultHelpers(), createMetrics(), logger.Named("manager"))
	require.NoError(t, err)
	enq, err := notifications.NewStoreEnqueuer(cfg, store, defaultHelpers(), logger.Named("enqueuer"), quartz.NewReal())
	require.NoError(t, err)

	user := createSampleUser(t, store)

	// WHEN: a few notifications are enqueued which will all succeed
	var msgs []string
	for i := 0; i < msgCount; i++ {
		ids, err := enq.Enqueue(ctx, user.ID, notifications.TemplateWorkspaceDeleted,
			map[string]string{"type": "success", "index": fmt.Sprintf("%d", i)}, "test")
		require.NoError(t, err)
		require.Len(t, ids, 2)
		msgs = append(msgs, ids[0].String(), ids[1].String())
	}

	mgr.Run(mgrCtx)

	// THEN:

	// Wait for the messages to be acquired
	<-noopInterceptor.acquiredChan
	// Then cancel the context, forcing the notification manager to shutdown ungracefully (simulating a crash); leaving messages in "leased" status.
	cancelManagerCtx()

	// Fetch any messages currently in "leased" status, and verify that they're exactly the ones we enqueued.
	leased, err := store.GetNotificationMessagesByStatus(ctx, database.GetNotificationMessagesByStatusParams{
		Status: database.NotificationMessageStatusLeased,
		Limit:  msgCount * 2,
	})
	require.NoError(t, err)