Method process

coderd/notifications/notifier.go:158–204

process is responsible for coordinating the retrieval, processing, and delivery of messages. Messages are dispatched concurrently, but they may block when success/failure channels are full. NOTE: it is _possible_ that these goroutines could block for long enough to exceed CODER_NOTIFICATIONS_DISPAT

(ctx context.Context, success chan<- dispatchResult, failure chan<- dispatchResult)

// resulting in a failed attempt for each notification when their contexts are canceled; this is not possible with the
// default configurations but could be brought about by an operator tuning things incorrectly.
func (n *notifier) process(ctx context.Context, success chan<- dispatchResult, failure chan<- dispatchResult) error {
	msgs, err := n.fetch(ctx)
	if err != nil {
		return xerrors.Errorf("fetch messages: %w", err)
	}

	n.log.Debug(ctx, "dequeued messages", slog.F("count", len(msgs)))

	if len(msgs) == 0 {
		return nil
	}

	var eg errgroup.Group
	for _, msg := range msgs {
		// If a notification template has been disabled by the user after a notification was enqueued, mark it as inhibited
		if msg.Disabled {
			failure <- n.newInhibitedDispatch(msg)
			continue
		}

		// A message failing to be prepared correctly should not affect other messages.
		deliverFn, err := n.prepare(ctx, msg)
		if err != nil {
			if database.IsQueryCanceledError(err) {
				n.log.Debug(ctx, "dispatcher construction canceled", slog.F("msg_id", msg.ID), slog.Error(err))
			} else {
				n.log.Error(ctx, "dispatcher construction failed", slog.F("msg_id", msg.ID), slog.Error(err))
			}
			failure <- n.newFailedDispatch(msg, err, xerrors.Is(err, decorateHelpersError{}))
			n.metrics.PendingUpdates.Set(float64(len(success) + len(failure)))
			continue
		}

		eg.Go(func() error {
			// Dispatch must only return an error for exceptional cases, NOT for failed messages.
			return n.deliver(ctx, msg, deliverFn, success, failure)
		})
	}

	if err = eg.Wait(); err != nil {
		n.log.Debug(ctx, "dispatch failed", slog.Error(err))
		return xerrors.Errorf("dispatch failed: %w", err)
	}

	n.log.Debug(ctx, "batch completed", slog.F("count", len(msgs)))
	return nil
}

// fetch retrieves messages from the queue by "acquiring a lease" whereby this notifier is the exclusive handler of these
// messages until they are dispatched - or until the lease expires (in exceptional cases).