feat: replace message_ttl with static max retry count (ory#2638)

This PR replaces the `courier.message_ttl` configuration option with a `courier.message_retries` option to limit how often the sending of a message is retried before it is marked as `abandoned`. 

BREAKING CHANGES: This is a breaking change, as it removes the `courier.message_ttl` config key and replaces it with a counter `courier.message_retries`.

Closes ory#402
Closes ory#1598

.schemastore/config.schema.json

@@ -1471,15 +1471,13 @@
  "/conf/courier-templates"
  ]
  },
- "message_ttl": {
- "description": "Defines a Time-To-Live for courier messages that could not be delivered. After the defined TTL has expired for a message that message is abandoned.",
- "type": "string",
- "pattern": "^([0-9]+(ns|us|ms|s|m|h))+$",
- "default": "1h",
+ "message_retries": {
+ "description": "Defines the maximum number of times the sending of a message is retried after it failed before it is marked as abandoned",
+ "type": "integer",
+ "default": 5,
  "examples": [
- "1h",
- "1m",
- "1s"
+ 10,
+ 60
  ]
  },
  "smtp": {

courier/courier_dispatcher.go

@@ -2,12 +2,37 @@ package courier
 
 import (
  "context"
- "time"
 
  "github.com/pkg/errors"
 )
 
 func (c *courier) DispatchMessage(ctx context.Context, msg Message) error {
+ maxRetries := c.deps.CourierConfig(ctx).CourierMessageRetries()
+
+ if msg.SendCount > maxRetries {
+ if err := c.deps.CourierPersister().SetMessageStatus(ctx, msg.ID, MessageStatusAbandoned); err != nil {
+ c.deps.Logger().
+ WithError(err).
+ WithField("message_id", msg.ID).
+ Error(`Unable to reset the retried message's status to "abandoned".`)
+ return err
+ }
+
+ // Skip the message
+ c.deps.Logger().
+ WithField("message_id", msg.ID).
+ Warnf(`Message was abandoned because it did not deliver after %d attempts`, msg.SendCount)
+ return nil
+ }
+
+ if err := c.deps.CourierPersister().IncrementMessageSendCount(ctx, msg.ID); err != nil {
+ c.deps.Logger().
+ WithError(err).
+ WithField("message_id", msg.ID).
+ Error(`Unable to increment the message's "send_count" field`)
+ return err
+ }
+
  switch msg.Type {
  case MessageTypeEmail:
  if err := c.dispatchEmail(ctx, msg); err != nil {
@@ -48,35 +73,22 @@ func (c *courier) DispatchQueue(ctx context.Context) error {
  return err
  }
 
- ttl := c.deps.CourierConfig(ctx).CourierMessageTTL()
-
  for k, msg := range messages {
- if time.Now().After(msg.CreatedAt.Add(ttl)) {
- if err := c.deps.CourierPersister().SetMessageStatus(ctx, msg.ID, MessageStatusAbandoned); err != nil {
- if c.failOnError {
- return err
- }
- c.deps.Logger().
- WithError(err).
- WithField("message_id", msg.ID).
- Error(`Unable to reset the timed out message's status to "abandoned".`)
- }
- } else {
- if err := c.DispatchMessage(ctx, msg); err != nil {
- for _, replace := range messages[k:] {
- if err := c.deps.CourierPersister().SetMessageStatus(ctx, replace.ID, MessageStatusQueued); err != nil {
- if c.failOnError {
- return err
- }
- c.deps.Logger().
- WithError(err).
- WithField("message_id", replace.ID).
- Error(`Unable to reset the failed message's status to "queued".`)
+ if err := c.DispatchMessage(ctx, msg); err != nil {
+
+ for _, replace := range messages[k:] {
+ if err := c.deps.CourierPersister().SetMessageStatus(ctx, replace.ID, MessageStatusQueued); err != nil {
+ if c.failOnError {
+ return err
  }
+ c.deps.Logger().
+ WithError(err).
+ WithField("message_id", replace.ID).
+ Error(`Unable to reset the failed message's status to "queued".`)
  }
-
- return err
  }
+
+ return err
  }
  }
 

courier/courier_dispatcher_test.go

@@ -3,28 +3,75 @@ package courier_test
 import (
  "context"
  "testing"
- "time"
 
  "github.com/gofrs/uuid"
- "github.com/sirupsen/logrus"
  "github.com/stretchr/testify/require"
 
  "github.com/ory/kratos/courier"
+ "github.com/ory/kratos/courier/template"
  templates "github.com/ory/kratos/courier/template/email"
  "github.com/ory/kratos/driver/config"
  "github.com/ory/kratos/internal"
 )
 
-func TestMessageTTL(t *testing.T) {
- if testing.Short() {
- t.SkipNow()
- }
+func queueNewMessage(t *testing.T, ctx context.Context, c courier.Courier, d template.Dependencies) uuid.UUID {
+ t.Helper()
+ id, err := c.QueueEmail(ctx, templates.NewTestStub(d, &templates.TestStubModel{
+ To: "[email protected]",
+ Subject: "test-subject-1",
+ Body: "test-body-1",
+ }))
+ require.NoError(t, err)
+ return id
+}
+
+func TestDispatchMessageWithInvalidSMTP(t *testing.T) {
  ctx := context.Background()
 
  conf, reg := internal.NewRegistryDefaultWithDSN(t, "")
- conf.MustSet(config.ViperKeyCourierMessageTTL, 1*time.Nanosecond)
+ conf.MustSet(config.ViperKeyCourierMessageRetries, 5)
+ conf.MustSet(config.ViperKeyCourierSMTPURL, "http://foo.url")
 
- reg.Logger().Level = logrus.TraceLevel
+ ctx, cancel := context.WithCancel(ctx)
+ defer cancel()
+
+ c := reg.Courier(ctx)
+
+ t.Run("case=failed sending", func(t *testing.T) {
+ id := queueNewMessage(t, ctx, c, reg)
+ message, err := reg.CourierPersister().LatestQueuedMessage(ctx)
+ require.NoError(t, err)
+ require.Equal(t, id, message.ID)
+
+ err = c.DispatchMessage(ctx, *message)
+ // sending the email fails, because there is no SMTP server at foo.url
+ require.Error(t, err)
+
+ messages, err := reg.CourierPersister().NextMessages(ctx, 10)
+ require.Len(t, messages, 1)
+ })
+
+ t.Run("case=max retries reached", func(t *testing.T) {
+ id := queueNewMessage(t, ctx, c, reg)
+ message, err := reg.CourierPersister().LatestQueuedMessage(ctx)
+ require.NoError(t, err)
+ require.Equal(t, id, message.ID)
+ message.SendCount = 6
+
+ err = c.DispatchMessage(ctx, *message)
+ require.NoError(t, err)
+
+ messages, err := reg.CourierPersister().NextMessages(ctx, 1)
+ require.Empty(t, messages)
+ })
+
+}
+
+func TestDispatchMessage2(t *testing.T) {
+ ctx := context.Background()
+
+ conf, reg := internal.NewRegistryDefaultWithDSN(t, "")
+ conf.MustSet(config.ViperKeyCourierMessageRetries, 1)
 
  c := reg.Courier(ctx)
 
@@ -39,9 +86,17 @@ func TestMessageTTL(t *testing.T) {
  require.NoError(t, err)
  require.NotEqual(t, uuid.Nil, id)
 
- c.DispatchQueue(ctx)
+ // Fails to deliver the first time
+ err = c.DispatchQueue(ctx)
+ require.Error(t, err)
+
+ // Retry once, as we set above - still fails
+ err = c.DispatchQueue(ctx)
+ require.Error(t, err)
 
- time.Sleep(1 * time.Second)
+ // Now it has been retried once, which means 2 > 1 is true and it is no longer tried
+ err = c.DispatchQueue(ctx)
+ require.NoError(t, err)
 
  var message courier.Message
  err = reg.Persister().GetConnection(ctx).

courier/message.go

@@ -27,20 +27,21 @@ const (
 
 // swagger:ignore
 type Message struct {
- ID uuid.UUID `json:"-" faker:"-" db:"id"`
- NID uuid.UUID `json:"-"  faker:"-" db:"nid"`
- Status MessageStatus `json:"-" db:"status"`
- Type MessageType `json:"-" db:"type"`
- Recipient string `json:"-" db:"recipient"`
- Body string `json:"-" db:"body"`
- Subject string `json:"-" db:"subject"`
- TemplateType TemplateType `json:"-" db:"template_type"`
+ ID uuid.UUID `json:"id" faker:"-" db:"id"`
+ NID uuid.UUID `json:"-" faker:"-" db:"nid"`
+ Status MessageStatus `json:"status" db:"status"`
+ Type MessageType `json:"type" db:"type"`
+ Recipient string `json:"recipient" db:"recipient"`
+ Body string `json:"body" db:"body"`
+ Subject string `json:"subject" db:"subject"`
+ TemplateType TemplateType `json:"template_type" db:"template_type"`
  TemplateData []byte `json:"-" db:"template_data"`
+ SendCount int `json:"send_count" db:"send_count"`
 
  // CreatedAt is a helper struct field for gobuffalo.pop.
- CreatedAt time.Time `json:"-" faker:"-" db:"created_at"`
+ CreatedAt time.Time `json:"created_at" faker:"-" db:"created_at"`
  // UpdatedAt is a helper struct field for gobuffalo.pop.
- UpdatedAt time.Time `json:"-" faker:"-" db:"updated_at"`
+ UpdatedAt time.Time `json:"updated_at" faker:"-" db:"updated_at"`
 }
 
 func (m Message) TableName(ctx context.Context) string {

courier/persistence.go

@@ -18,6 +18,8 @@ type (
  SetMessageStatus(context.Context, uuid.UUID, MessageStatus) error
 
  LatestQueuedMessage(ctx context.Context) (*Message, error)
+
+ IncrementMessageSendCount(context.Context, uuid.UUID) error
  }
  PersistenceProvider interface {
  CourierPersister() Persister

courier/smtp.go

@@ -5,6 +5,7 @@ import (
  "crypto/tls"
  "encoding/json"
  "fmt"
+ "net/textproto"
  "strconv"
  "time"
 
@@ -199,6 +200,19 @@ func (c *courier) dispatchEmail(ctx context.Context, msg Message) error {
  // WithField("email_to", msg.Recipient).
  WithField("message_from", from).
  Error("Unable to send email using SMTP connection.")
+
+ var protoErr *textproto.Error
+ if containsProtoErr := errors.As(err, &protoErr); containsProtoErr && protoErr.Code >= 500 {
+ // See https://en.wikipedia.org/wiki/List_of_SMTP_server_return_codes
+ // If the SMTP server responds with 5xx, sending the message should not be retried (without changing something about the request)
+ if err := c.deps.CourierPersister().SetMessageStatus(ctx, msg.ID, MessageStatusAbandoned); err != nil {
+ c.deps.Logger().
+ WithError(err).
+ WithField("message_id", msg.ID).
+ Error(`Unable to reset the retried message's status to "abandoned".`)
+ return err
+ }
+ }
  return errors.WithStack(err)
  }
 

courier/test/persistence.go

@@ -95,6 +95,18 @@ func TestPersister(ctx context.Context, newNetworkUnlessExisting NetworkWrapper,
  require.ErrorIs(t, err, courier.ErrQueueEmpty)
  })
 
+ t.Run("case=incrementing send count", func(t *testing.T) {
+ originalSendCount := messages[0].SendCount
+ require.NoError(t, p.SetMessageStatus(ctx, messages[0].ID, courier.MessageStatusQueued))
+
+ require.NoError(t, p.IncrementMessageSendCount(ctx, messages[0].ID))
+ ms, err := p.NextMessages(ctx, 1)
+ require.NoError(t, err)
+ require.Len(t, ms, 1)
+ assert.Equal(t, messages[0].ID, ms[0].ID)
+ assert.Equal(t, originalSendCount+1, ms[0].SendCount)
+ })
+
  t.Run("case=network", func(t *testing.T) {
  id := x.NewUUID()
 

driver/config/config.go

@@ -74,7 +74,7 @@ const (
  ViperKeyCourierSMSRequestConfig = "courier.sms.request_config"
  ViperKeyCourierSMSEnabled = "courier.sms.enabled"
  ViperKeyCourierSMSFrom = "courier.sms.from"
- ViperKeyCourierMessageTTL  = "courier.message_ttl"
+ ViperKeyCourierMessageRetries = "courier.message_retries"
  ViperKeySecretsDefault = "secrets.default"
  ViperKeySecretsCookie = "secrets.cookie"
  ViperKeySecretsCipher = "secrets.cipher"
@@ -257,7 +257,7 @@ type (
  CourierTemplatesVerificationValid() *CourierEmailTemplate
  CourierTemplatesRecoveryInvalid() *CourierEmailTemplate
  CourierTemplatesRecoveryValid() *CourierEmailTemplate
- CourierMessageTTL() time.Duration
+ CourierMessageRetries() int
  }
 )
 
@@ -948,8 +948,8 @@ func (p *Config) CourierTemplatesRecoveryValid() *CourierEmailTemplate {
  return p.CourierTemplatesHelper(ViperKeyCourierTemplatesRecoveryValidEmail)
 }
 
-func (p *Config) CourierMessageTTL() time.Duration {
- return p.p.DurationF(ViperKeyCourierMessageTTL, time.Hour)
+func (p *Config) CourierMessageRetries() int {
+ return p.p.IntF(ViperKeyCourierMessageRetries, 5)
 }
 
 func (p *Config) CourierSMTPHeaders() map[string]string {

driver/config/config_test.go

@@ -1110,13 +1110,13 @@ func TestCourierMessageTTL(t *testing.T) {
 
  t.Run("case=configs set", func(t *testing.T) {
  conf, _ := config.New(ctx, logrusx.New("", ""), os.Stderr,
- configx.WithConfigFiles("stub/.kratos.courier.messageTTL.yaml"), configx.SkipValidation())
- assert.Equal(t, conf.CourierMessageTTL(), time.Duration(5*time.Minute))
+ configx.WithConfigFiles("stub/.kratos.courier.message_retries.yaml"), configx.SkipValidation())
+ assert.Equal(t, conf.CourierMessageRetries(), 10)
  })
 
  t.Run("case=defaults", func(t *testing.T) {
  conf, _ := config.New(ctx, logrusx.New("", ""), os.Stderr, configx.SkipValidation())
- assert.Equal(t, conf.CourierMessageTTL(), time.Duration(1*time.Hour))
+ assert.Equal(t, conf.CourierMessageRetries(), 5)
  })
 }
 

driver/config/stub/.kratos.courier.message_retries.yaml

@@ -0,0 +1,2 @@
+courier:
+ message_retries: 10

embedx/config.schema.json

@@ -1549,15 +1549,13 @@
  "/conf/courier-templates"
  ]
  },
- "message_ttl": {
- "description": "Defines a Time-To-Live for courier messages that could not be delivered. After the defined TTL has expired for a message that message is abandoned.",
- "type": "string",
- "pattern": "^([0-9]+(ns|us|ms|s|m|h))+$",
- "default": "1h",
+ "message_retries": {
+ "description": "Defines the maximum number of times the sending of a message is retried after it failed before it is marked as abandoned",
+ "type": "integer",
+ "default": 5,
  "examples": [
- "1h",
- "1m",
- "1s"
+ 10,
+ 60
  ]
  },
  "smtp": {