MDI240 | handle aws send email throttling requeing the email in status READY

Author: emidiocruciani

docs/error-handling.md

@@ -41,6 +41,12 @@ Quando l'invio SMTP fallisce:
 - Stato: `FAILED`
 - Reason: Messaggio di errore originale dall'SMTP client
 
+### SMTP Throttling (454)
+Quando l'invio SMTP fallisce con codice `454` (throttling):
+- Stato: ritorna a `READY`
+- Record `PROCESSING` rimosso da `email_statuses`
+- Reason: vuota
+
 ### Callback Failed
 Quando il callback HTTP fallisce dopo tutti i retry:
 - Stato rimane: `CALLING-SENT-CALLBACK` o `CALLING-FAILED-CALLBACK`

internal/outbox/outbox.go

@@ -237,6 +237,60 @@ func (o *Outbox) Update(ctx context.Context, id string, status string, errorReas
 	return err
 }
 
+// Requeue updates the email from PROCESSING back to READY and removes the PROCESSING history row.
+// The operation is executed within a transaction with retry logic for transient errors.
+// Note: ttl parameter is ignored for MySQL.
+func (o *Outbox) Requeue(ctx context.Context, id string, _ *int64) error {
+	updateQuery := `
+		UPDATE emails
+		SET status = ?, reason = ?, version = version + 1
+		WHERE id = ? AND status = ?
+	`
+	deleteProcessingQuery := `
+		DELETE FROM email_statuses
+		WHERE email_id = ? AND status = ?
+		ORDER BY id DESC
+		LIMIT 1
+	`
+
+	var err error
+	for attempt := range maxAttempts {
+		err = o.executeInTransaction(ctx, func(tx *sql.Tx) error {
+			result, execErr := tx.ExecContext(ctx, updateQuery, StatusReady, "", id, StatusProcessing)
+			if execErr != nil {
+				return execErr
+			}
+
+			affected, affErr := result.RowsAffected()
+			if affErr != nil {
+				return affErr
+			}
+
+			if affected == 0 {
+				return ErrLockNotAcquired
+			}
+
+			_, delErr := tx.ExecContext(ctx, deleteProcessingQuery, id, StatusProcessing)
+			return delErr
+		})
+
+		if err == nil || !o.shouldRetryMySQL(err) {
+			return err
+		}
+
+		sleep := o.backoffDuration(attempt)
+		timer := time.NewTimer(sleep)
+		select {
+		case <-ctx.Done():
+			timer.Stop()
+			return ctx.Err()
+		case <-timer.C:
+		}
+	}
+
+	return err
+}
+
 // Ready updates the email to READY status with the eml file path.
 // Expected from status is INTAKING.
 // The operation is executed within a transaction with retry logic for transient errors.

internal/pipeline/interface.go

@@ -9,4 +9,5 @@ type outboxService interface {
 	Query(ctx context.Context, status string, limit int) ([]outbox.Email, error)
 	Update(ctx context.Context, id string, status string, errorReason string, ttl *int64) error
 	Ready(ctx context.Context, id string, emlFilePath string, ttl *int64) error
+	Requeue(ctx context.Context, id string, ttl *int64) error
 }

internal/pipeline/sender.go

@@ -2,8 +2,10 @@ package pipeline
 
 import (
 	"context"
+	"errors"
 	"fmt"
 	"log/slog"
+	"net/textproto"
 	"sync"
 
 	"mailculator-processor/internal/email"
@@ -59,8 +61,15 @@ func (p *MainSenderPipeline) Process(ctx context.Context) {
 			}
 
 			if err = p.client.Send(payload, p.attachmentsBasePath); err != nil {
-				logger.Error(fmt.Sprintf("failed to send, error: %v", err))
-				p.handle(context.Background(), logger, outboxEmail.Id, outbox.StatusFailed, err.Error(), outboxEmail.TTL)
+				if isSMTPThrottling(err) {
+					logger.Warn(fmt.Sprintf("smtp throttling, requeueing: %v", err))
+					if requeueErr := p.outbox.Requeue(context.Background(), outboxEmail.Id, outboxEmail.TTL); requeueErr != nil {
+						logger.Error(fmt.Sprintf("error requeueing email, error: %v", requeueErr))
+					}
+				} else {
+					logger.Error(fmt.Sprintf("failed to send, error: %v", err))
+					p.handle(context.Background(), logger, outboxEmail.Id, outbox.StatusFailed, err.Error(), outboxEmail.TTL)
+				}
 			} else {
 				logger.Info("successfully sent")
 				p.handle(context.Background(), logger, outboxEmail.Id, outbox.StatusSent, "", outboxEmail.TTL)
@@ -77,3 +86,16 @@ func (p *MainSenderPipeline) handle(ctx context.Context, logger *slog.Logger, em
 		logger.Error(msg)
 	}
 }
+
+func isSMTPThrottling(err error) bool {
+	if err == nil {
+		return false
+	}
+
+	var smtpErr *textproto.Error
+	if errors.As(err, &smtpErr) {
+		return smtpErr.Code == 454
+	}
+
+	return false
+}

internal/pipeline/sender_test.go

@@ -6,6 +6,7 @@ import (
 	"context"
 	"encoding/json"
 	"errors"
+	"net/textproto"
 	"os"
 	"strings"
 	"testing"
@@ -123,6 +124,25 @@ func TestSendEmailError(t *testing.T) {
 	)
 }
 
+func TestSendEmailThrottlingRequeue(t *testing.T) {
+	payloadFile := createPayloadFile(t)
+	buf, logger := mocks.NewLoggerMock()
+	outboxServiceMock := mocks.NewOutboxMock(
+		mocks.Email(outbox.Email{Id: "1", Status: "", PayloadFilePath: payloadFile}),
+	)
+	senderServiceMock := newSenderMock(&textproto.Error{Code: 454, Msg: "Throttling failure"})
+	sender := MainSenderPipeline{outbox: outboxServiceMock, client: senderServiceMock, attachmentsBasePath: "/base/path/", logger: logger}
+
+	sender.Process(context.TODO())
+
+	assert.Equal(t, 0, senderServiceMock.sendMethodCounter)
+	assert.Equal(t, "requeue", outboxServiceMock.LastMethod())
+	assert.Equal(t,
+		"level=INFO msg=\"processing outbox 1\"\nlevel=WARN msg=\"smtp throttling, requeueing: 454 Throttling failure\" outbox=1",
+		strings.TrimSpace(buf.String()),
+	)
+}
+
 func TestHandleUpdateError(t *testing.T) {
 	payloadFile := createPayloadFile(t)
 	buf, logger := mocks.NewLoggerMock()

internal/testutils/mocks/outbox.go

@@ -10,7 +10,11 @@ type OutboxMock struct {
 	updateMethodError     error
 	updateMethodCall      int
 	updateMethodFailsCall int
+	requeueMethodError     error
+	requeueMethodCall      int
+	requeueMethodFailsCall int
 	email                 outbox.Email
+	lastMethod            string
 }
 
 type OutboxMockOptions func(*OutboxMock)
@@ -33,6 +37,18 @@ func UpdateMethodFailsCall(updateMethodFailsCall int) OutboxMockOptions {
 	}
 }
 
+func RequeueMethodError(requeueMethodError error) OutboxMockOptions {
+	return func(o *OutboxMock) {
+		o.requeueMethodError = requeueMethodError
+	}
+}
+
+func RequeueMethodFailsCall(requeueMethodFailsCall int) OutboxMockOptions {
+	return func(o *OutboxMock) {
+		o.requeueMethodFailsCall = requeueMethodFailsCall
+	}
+}
+
 func Email(email outbox.Email) OutboxMockOptions {
 	return func(o *OutboxMock) {
 		o.email = email
@@ -45,7 +61,11 @@ func NewOutboxMock(opts ...OutboxMockOptions) *OutboxMock {
 		updateMethodError:     nil,
 		updateMethodCall:      0,
 		updateMethodFailsCall: 1,
+		requeueMethodError:     nil,
+		requeueMethodCall:      0,
+		requeueMethodFailsCall: 1,
 		email:                 outbox.Email{},
+		lastMethod:            "",
 	}
 	for _, opt := range opts {
 		opt(o)
@@ -54,10 +74,12 @@ func NewOutboxMock(opts ...OutboxMockOptions) *OutboxMock {
 }
 
 func (m *OutboxMock) Query(ctx context.Context, status string, limit int) ([]outbox.Email, error) {
+	m.lastMethod = "query"
 	return []outbox.Email{m.email}, m.queryMethodError
 }
 
 func (m *OutboxMock) Update(ctx context.Context, id string, status string, errorReason string, ttl *int64) error {
+	m.lastMethod = "update"
 	m.updateMethodCall++
 	if m.updateMethodCall == m.updateMethodFailsCall {
 		return m.updateMethodError
@@ -66,9 +88,23 @@ func (m *OutboxMock) Update(ctx context.Context, id string, status string, error
 }
 
 func (m *OutboxMock) Ready(ctx context.Context, id string, emlFilePath string, ttl *int64) error {
+	m.lastMethod = "ready"
 	m.updateMethodCall++
 	if m.updateMethodCall == m.updateMethodFailsCall {
 		return m.updateMethodError
 	}
 	return nil
 }
+
+func (m *OutboxMock) Requeue(ctx context.Context, id string, ttl *int64) error {
+	m.lastMethod = "requeue"
+	m.requeueMethodCall++
+	if m.requeueMethodCall == m.requeueMethodFailsCall {
+		return m.requeueMethodError
+	}
+	return nil
+}
+
+func (m *OutboxMock) LastMethod() string {
+	return m.lastMethod
+}