Method deliver

coderd/notifications/notifier.go:266–322

deliver sends a given notification message via its defined method. This method *only* returns an error when a context error occurs; any other error is interpreted as a failure to deliver the notification and as such the message will be marked as failed (to later be optionally retried).

(ctx context.Context, msg database.AcquireNotificationMessagesRow, deliver dispatch.DeliveryFunc, success, failure chan<- dispatchResult)

// This method *only* returns an error when a context error occurs; any other error is interpreted as a failure to
// deliver the notification and as such the message will be marked as failed (to later be optionally retried).
func (n *notifier) deliver(ctx context.Context, msg database.AcquireNotificationMessagesRow, deliver dispatch.DeliveryFunc, success, failure chan<- dispatchResult) error {
	select {
	case <-ctx.Done():
		return ctx.Err()
	default:
	}

	ctx, cancel := context.WithTimeout(ctx, n.cfg.DispatchTimeout.Value())
	defer cancel()
	logger := n.log.With(slog.F("msg_id", msg.ID), slog.F("method", msg.Method), slog.F("attempt", msg.AttemptCount+1))

	if msg.AttemptCount > 0 {
		n.metrics.RetryCount.WithLabelValues(string(msg.Method), msg.TemplateID.String()).Inc()
	}

	n.metrics.InflightDispatches.WithLabelValues(string(msg.Method), msg.TemplateID.String()).Inc()
	n.metrics.QueuedSeconds.WithLabelValues(string(msg.Method)).Observe(msg.QueuedSeconds)

	start := n.clock.Now()
	retryable, err := deliver(ctx, msg.ID)

	n.metrics.DispatcherSendSeconds.WithLabelValues(string(msg.Method)).Observe(n.clock.Since(start).Seconds())
	n.metrics.InflightDispatches.WithLabelValues(string(msg.Method), msg.TemplateID.String()).Dec()

	if err != nil {
		// Don't try to accumulate message responses if the context has been canceled.
		//
		// This message's lease will expire in the store and will be requeued.
		// It's possible this will lead to a message being delivered more than once, and that is why Stop() is preferable
		// instead of canceling the context.
		//
		// In the case of backpressure (i.e. the success/failure channels are full because the database is slow),
		// we can't append any more updates to the channels otherwise this, too, will block.
		if xerrors.Is(err, context.Canceled) {
			return err
		}

		select {
		case <-ctx.Done():
			logger.Warn(context.Background(), "cannot record dispatch failure result", slog.Error(ctx.Err()))
			return ctx.Err()
		case failure <- n.newFailedDispatch(msg, err, retryable):
			logger.Warn(ctx, "message dispatch failed", slog.Error(err))
		}
	} else {
		select {
		case <-ctx.Done():
			logger.Warn(context.Background(), "cannot record dispatch success result", slog.Error(ctx.Err()))
			return ctx.Err()
		case success <- n.newSuccessfulDispatch(msg):
			logger.Debug(ctx, "message dispatch succeeded")
		}
	}
	n.metrics.PendingUpdates.Set(float64(len(success) + len(failure)))

	return nil
}