From 9874ff12915721d45f7b81fa3e1e98f586de2ed0 Mon Sep 17 00:00:00 2001 From: Matthew Meszaros Date: Sun, 23 Aug 2026 10:18:24 -0700 Subject: [PATCH] feat: make the worker answer every SEND_EMAIL it acks with exactly one per-task EMAIL_FAILED or EMAIL_SENT, raising the typed account events (auth, disabled, rate limited, server error) in addition with the EmailErrorEvent body the consumer expects instead of a SendEmailResult it could not parse; a mailbox that is not loaded yet or a storage blip is left for a few bus redeliveries before being reported, using the new Message.Attempt and Redelivers fields the NATS and Kafka buses now fill in, and an SMTP INVALID_CREDENTIALS send failure is classified as an auth error --- internal/app/worker/event_send_email.go | 49 ++++++++++++++++++----- internal/app/worker/receive.go | 17 +++++++- internal/app/worker/wmail/send.go | 2 +- internal/infrastructure/eventbus/bus.go | 9 +++++ internal/infrastructure/eventbus/kafka.go | 1 + internal/infrastructure/eventbus/nats.go | 12 ++++-- 6 files changed, 74 insertions(+), 16 deletions(-) diff --git a/internal/app/worker/event_send_email.go b/internal/app/worker/event_send_email.go index d16def42..5b253e5f 100644 --- a/internal/app/worker/event_send_email.go +++ b/internal/app/worker/event_send_email.go @@ -3,6 +3,7 @@ package worker import ( "bytes" "context" + "errors" "fmt" "io" "time" @@ -29,9 +30,13 @@ func (w *WorkerService) HandleSendEmail(ctx context.Context, sendEmail models.Se w.mailManager.RUnlock() if !exists { + // The mailbox is not loaded here: it is still being added (its + // ADD_EMAIL is queued behind this send), or this worker restarted and + // the reconciler has not re-shipped it yet. Leave the send for + // redelivery a few times so a queued ADD_EMAIL gets processed first, + // then report the failure so the control plane retries the step. err := fmt.Errorf("email account %s not found in worker", sendEmail.EmailID.String()) - w.sendEmailFailure(sendEmail.TaskID, sendEmail.EmailID, mail, err.Error()) - return err + return w.failSend(ctx, sendEmail, err.Error(), true) } // Decrypt subject @@ -50,8 +55,7 @@ func (w *WorkerService) HandleSendEmail(ctx context.Context, sendEmail models.Se bodyPlain, bodyHTML, attachmentRefs, err := w.fetchEmailBody(ctx, sendEmail.OrgID, sendEmail.BodyS3Key) if err != nil { log.Error().Err(err).Str("s3_key", sendEmail.BodyS3Key).Msg("Failed to fetch email body from S3") - w.sendEmailFailure(sendEmail.TaskID, sendEmail.EmailID, mail, fmt.Sprintf("failed to fetch email body: %v", err)) - return err + return w.failSend(ctx, sendEmail, fmt.Sprintf("failed to fetch email body: %v", err), true) } // Fetch each attachment's bytes from object storage by key. A fetch failure @@ -59,8 +63,7 @@ func (w *WorkerService) HandleSendEmail(ctx context.Context, sendEmail models.Se attachments, err := w.fetchAttachments(ctx, attachmentRefs) if err != nil { log.Error().Err(err).Str("task_id", sendEmail.TaskID.String()).Msg("Failed to fetch attachment bytes from S3") - w.sendEmailFailure(sendEmail.TaskID, sendEmail.EmailID, mail, fmt.Sprintf("failed to fetch attachment: %v", err)) - return err + return w.failSend(ctx, sendEmail, fmt.Sprintf("failed to fetch attachment: %v", err), true) } // Use unified Send method @@ -220,7 +223,30 @@ func (w *WorkerService) sendEmailSuccess(taskID uuid.UUID, messageID, providerMs } } -// sendEmailError sends a structured error result back to the jobs service +// sendNotLoadedRedeliveries is how many bus deliveries a send may burn waiting +// for a transient condition (mailbox not loaded yet, object storage blip) +// before the worker reports it failed. Each redelivery is a second apart, and +// the bus stops redelivering at ten, so this stays well inside that. +const sendNotLoadedRedeliveries = 5 + +// failSend reports a send the worker could not attempt. A retryable condition +// is first left for bus redelivery (returning an error naks the message) so a +// queued ADD_EMAIL or a storage blip can clear; once the redeliveries are +// spent, when the bus does not redeliver, or when the condition is not +// retryable, the failure result is produced and the message is acked, handing +// the retry to the control plane. +func (w *WorkerService) failSend(ctx context.Context, sendEmail models.SendEmail, reason string, retryable bool) error { + if d := deliveryOf(ctx); retryable && d.redelivers && d.attempt < sendNotLoadedRedeliveries { + return errors.New(reason) + } + w.sendEmailFailure(sendEmail.TaskID, sendEmail.EmailID, nil, reason) + return nil +} + +// sendEmailError reports a failed send attempt. The per-task result is always +// an EMAIL_FAILED so the control plane has one result channel to walk the +// send back on; account-level conditions (auth, disabled, rate limit, server +// error) additionally raise their own typed event carrying the full context. func (w *WorkerService) sendEmailError(taskID uuid.UUID, emailID uuid.UUID, mail *wmail.WMail, mailErr *errx.MailError) { // Determine the appropriate event type based on error eventType := wmail.DetermineErrorEventType(mailErr) @@ -236,14 +262,15 @@ func (w *WorkerService) sendEmailError(taskID uuid.UUID, emailID uuid.UUID, mail SentAt: time.Now(), } - if err := w.Produce(eventType, taskID.String(), result); err != nil { - log.Error().Err(err).Str("task_id", taskID.String()).Msg("Failed to produce email error event") + if err := w.Produce(models.JobEventTypeEmailFailed, taskID.String(), result); err != nil { + log.Error().Err(err).Str("task_id", taskID.String()).Msg("Failed to produce email failed event") } - // For critical auth/disabled errors, also send a separate error event with full context + // Account-level conditions also raise their typed event with full context if eventType == models.JobEventTypeEmailAuthError || eventType == models.JobEventTypeEmailDisabled || - eventType == models.JobEventTypeEmailRateLimited { + eventType == models.JobEventTypeEmailRateLimited || + eventType == models.JobEventTypeEmailServerError { userInfo := mailErr.GetUserErrorInfo() errorEvent := models.EmailErrorEvent{ diff --git a/internal/app/worker/receive.go b/internal/app/worker/receive.go index cbf1a7db..7b69f75d 100644 --- a/internal/app/worker/receive.go +++ b/internal/app/worker/receive.go @@ -8,6 +8,21 @@ import ( "github.com/warmbly/warmbly/internal/models" ) +type deliveryKey struct{} + +// delivery is what a handler may know about the bus message it is processing. +type delivery struct { + attempt int // 1-based delivery count; 0 when unknown + redelivers bool // a handler error gets the message delivered again +} + +// deliveryOf returns the bus delivery details for the message a handler is +// processing (zero value when the handler was not invoked from Receive). +func deliveryOf(ctx context.Context) delivery { + d, _ := ctx.Value(deliveryKey{}).(delivery) + return d +} + // Receive is the eventbus.Handler that drives the worker's event loop. It // decodes the wire payload via the injected codec.Codec and dispatches to // HandleEvent. @@ -17,7 +32,7 @@ func (w *WorkerService) Receive(ctx context.Context, msg eventbus.Message) error return err } - hctx, cancel := context.WithTimeout(ctx, 30*time.Second) + hctx, cancel := context.WithTimeout(context.WithValue(ctx, deliveryKey{}, delivery{attempt: msg.Attempt, redelivers: msg.Redelivers}), 30*time.Second) defer cancel() return w.HandleEvent(hctx, &event) } diff --git a/internal/app/worker/wmail/send.go b/internal/app/worker/wmail/send.go index 69123bab..6187bb61 100644 --- a/internal/app/worker/wmail/send.go +++ b/internal/app/worker/wmail/send.go @@ -422,7 +422,7 @@ func DetermineErrorEventType(err *errx.MailError) models.JobEventType { } switch err.Code { - case errx.MailErrorCodeGoogleAuth, errx.MailErrorCodeAuthenticationFailed: + case errx.MailErrorCodeGoogleAuth, errx.MailErrorCodeAuthenticationFailed, errx.MailErrorCodeInvalidCredentials: return models.JobEventTypeEmailAuthError case errx.MailErrorCodeAccountSuspended, errx.MailErrorCodeAuthorizationFailed: diff --git a/internal/infrastructure/eventbus/bus.go b/internal/infrastructure/eventbus/bus.go index 1a273b52..359aea71 100644 --- a/internal/infrastructure/eventbus/bus.go +++ b/internal/infrastructure/eventbus/bus.go @@ -76,6 +76,15 @@ type Message struct { Topic string Key string Payload []byte + // Attempt is the 1-based delivery count of this message (NATS reports it + // from the consumer's redelivery metadata; Kafka always reports 1). A + // handler that retries by returning an error can read it to know when the + // broker is about to stop redelivering and give up cleanly instead. + Attempt int + // Redelivers is true when a handler error leaves the message for another + // delivery (NATS). Kafka commits regardless, so a handler must not count + // on a retry there and should finish what it can on this delivery. + Redelivers bool } // Subject normalises a topic name to the dot-separated form by replacing any diff --git a/internal/infrastructure/eventbus/kafka.go b/internal/infrastructure/eventbus/kafka.go index 8303f517..479dc2a2 100644 --- a/internal/infrastructure/eventbus/kafka.go +++ b/internal/infrastructure/eventbus/kafka.go @@ -140,6 +140,7 @@ func (b *KafkaBus) Subscribe(ctx context.Context, topics []string, group string, Topic: topic, Key: string(msg.Key), Payload: msg.Value, + Attempt: 1, }); err != nil { log.Error().Err(err).Str("topic", topic).Msg("eventbus kafka handler error") return err diff --git a/internal/infrastructure/eventbus/nats.go b/internal/infrastructure/eventbus/nats.go index 490d0fbc..87a8e1f3 100644 --- a/internal/infrastructure/eventbus/nats.go +++ b/internal/infrastructure/eventbus/nats.go @@ -249,10 +249,16 @@ func (b *NATSBus) Subscribe(ctx context.Context, topics []string, group string, // on the exact separator and should use Subject() if it needs to // compare against a Kafka-style topic name. topic := strings.TrimPrefix(m.Subject(), b.prefix+".") + attempt := 1 + if meta, merr := m.Metadata(); merr == nil && meta.NumDelivered > 0 { + attempt = int(meta.NumDelivered) + } if err := invokeHandler(hctx, handler, Message{ - Topic: topic, - Key: key, - Payload: m.Data(), + Topic: topic, + Key: key, + Payload: m.Data(), + Attempt: attempt, + Redelivers: true, }); err != nil { log.Error().Err(err).Str("subject", m.Subject()).Msg("eventbus nats handler error") // Nak with a short delay so transient errors don't hot-loop.