first draft for concurrency using an approach at the database level

readme.md

@@ -74,11 +74,19 @@ public class App {
 
 The following configuration options are available:
 
-| Name                                   | Default     | Description                                                           |
-|----------------------------------------|-------------|-----------------------------------------------------------------------|
-| `lib.emailservice.migrations.enabled`  | true        | Enables database migrations.                                          |
-| `lib.emailservice.scheduling.enabled`  | true        | Enables the scheduler sending the emails.                             |
-| `lib.emailservice.scheduling.interval` | 30000       | Specifies the milliseconds delay between runs of the scheduler.       |
+| Name                                                                    | Default | Description                                                                                                            |
+|-------------------------------------------------------------------------|---------|------------------------------------------------------------------------------------------------------------------------|
+| `aboutbits.emailservice.migrations.enabled`                             | true    | Enables database migrations.                                                                                           |
+| `aboutbits.emailservice.scheduling.enabled`                             | true    | Enables the scheduler sending the emails.                                                                              |
+| `aboutbits.emailservice.scheduling.cleanup.enabled`                     | true    | Enables cleanup of attachment files after sending.                                                                     |
+| `aboutbits.emailservice.scheduling.interval`                            | 30000   | Milliseconds delay between runs of the scheduler.                                                                      |
+| `aboutbits.emailservice.scheduling.stuck-sending-recovery-threshold`    | PT30M   | How long an email may stay in `SENDING` before being considered abandoned (crashed pod) and eligible to be re-claimed. Must comfortably exceed the worst-case SMTP send duration: JavaMail's default connect/read/write timeouts are infinite, so configure `spring.mail.properties.mail.smtp.connectiontimeout`, `spring.mail.properties.mail.smtp.timeout` and `spring.mail.properties.mail.smtp.writetimeout` well below this threshold, otherwise a slow in-flight send can be re-claimed by another pod and delivered twice. |
+| `aboutbits.emailservice.scheduling.max-attempts`                        | 3       | Maximum number of send attempts before an email is marked as `ERROR`. Applies only to the scheduled retry loop. Failed attempts are retried with exponential backoff (`scheduling.interval` × 2^attempts); with the defaults a persistently failing email runs attempt 1 → +60s → attempt 2 → +120s → attempt 3 → `ERROR`. |
+
+## Multi-pod deployments
+
+The scheduler is safe to run on every pod concurrently: the database arbitrates which pod sends each email.
+Delivery is at-least-once - if a pod crashes after the SMTP server accepted the message but before the result was persisted, the email may be sent again on recovery.
 
 ## Local development:
 

src/main/java/it/aboutbits/springboot/emailservice/EmailServiceConfiguration.java

@@ -14,15 +14,17 @@
 import it.aboutbits.springboot.emailservice.lib.application.SendScheduledEmails;
 import it.aboutbits.springboot.emailservice.lib.application.UnavailableAttachmentDataSource;
 import it.aboutbits.springboot.emailservice.lib.jpa.EmailRepository;
-import jakarta.persistence.EntityManager;
 import org.jspecify.annotations.NullMarked;
+import org.springframework.beans.factory.annotation.Value;
 import org.springframework.boot.autoconfigure.AutoConfigurationPackage;
 import org.springframework.boot.autoconfigure.condition.ConditionalOnMissingBean;
 import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
 import org.springframework.context.annotation.Bean;
 import org.springframework.jdbc.core.JdbcTemplate;
 import org.springframework.mail.javamail.JavaMailSender;
+import org.springframework.transaction.PlatformTransactionManager;
 
+import java.time.Duration;
 import java.util.List;
 
 @AutoConfigurationPackage
@@ -45,24 +47,49 @@ public EmailMapper emailMapper() {
     }
 
     @Bean
-    public QueryEmail queryEmail(EmailRepository emailRepository, EmailMapper emailMapper, EntityManager entityManager) {
-        return new QueryEmail(emailRepository, emailMapper, entityManager);
+    public QueryEmail queryEmail(EmailRepository emailRepository, EmailMapper emailMapper) {
+        return new QueryEmail(emailRepository, emailMapper);
     }
 
     @Bean
-    public ManageEmail manageEmail(EmailRepository emailRepository, JavaMailSender javaMailSender, AttachmentDataSource attachmentDataSource, EmailMapper emailMapper) {
-        return new ManageEmail(emailRepository, javaMailSender, attachmentDataSource, emailMapper);
+    public ManageEmail manageEmail(
+            EmailRepository emailRepository,
+            JavaMailSender javaMailSender,
+            AttachmentDataSource attachmentDataSource,
+            EmailMapper emailMapper,
+            @Value("${aboutbits.emailservice.scheduling.max-attempts:3}") int maxAttempts,
+            @Value("${aboutbits.emailservice.scheduling.interval:30000}") long schedulerIntervalMillis,
+            PlatformTransactionManager transactionManager
+    ) {
+        return new ManageEmail(
+                emailRepository,
+                javaMailSender,
+                attachmentDataSource,
+                emailMapper,
+                maxAttempts,
+                Duration.ofMillis(schedulerIntervalMillis),
+                transactionManager
+        );
     }
 
     @Bean
     @ConditionalOnProperty(value = "aboutbits.emailservice.scheduling.enabled", matchIfMissing = true)
-    public SendScheduledEmails sendScheduledEmails(QueryEmail queryEmail, ManageEmail manageEmail, List<EmailSchedulerCallback> callbacks) {
-        return new SendScheduledEmails(queryEmail, manageEmail, callbacks);
+    public SendScheduledEmails sendScheduledEmails(
+            QueryEmail queryEmail,
+            ManageEmail manageEmail,
+            List<EmailSchedulerCallback> callbacks,
+            @Value("${aboutbits.emailservice.scheduling.stuck-sending-recovery-threshold:PT30M}") Duration stuckSendingRecoveryThreshold
+    ) {
+        return new SendScheduledEmails(queryEmail, manageEmail, callbacks, stuckSendingRecoveryThreshold);
     }
 
     @Bean
     @ConditionalOnProperty(value = "aboutbits.emailservice.scheduling.cleanup.enabled", matchIfMissing = true)
-    public CleanupAttachmentFiles cleanupAttachments(QueryEmail queryEmail, ManageEmail manageEmail, List<AttachmentCleanerCallback> callbacks) {
+    public CleanupAttachmentFiles cleanupAttachments(
+            QueryEmail queryEmail,
+            ManageEmail manageEmail,
+            List<AttachmentCleanerCallback> callbacks
+    ) {
         return new CleanupAttachmentFiles(queryEmail, manageEmail, callbacks);
     }
 

src/main/java/it/aboutbits/springboot/emailservice/lib/EmailDto.java

@@ -32,10 +32,12 @@ public record EmailDto(
 
         OffsetDateTime scheduledAt,
         @Nullable
-        OffsetDateTime sentAt,
-
+        OffsetDateTime executionStartTime,
         @Nullable
-        OffsetDateTime errorAt,
+        OffsetDateTime executionEndTime,
+
+        int attempts,
+
         @Nullable
         String errorMessage,
 

src/main/java/it/aboutbits/springboot/emailservice/lib/EmailState.java

@@ -5,6 +5,7 @@
 @NullMarked
 public enum EmailState {
     PENDING,
+    SENDING,
     SENT,
     ERROR
 }

src/main/java/it/aboutbits/springboot/emailservice/lib/application/EmailMapper.java

@@ -6,6 +6,7 @@
 import org.jspecify.annotations.NullUnmarked;
 import org.mapstruct.AnnotateWith;
 import org.mapstruct.Mapper;
+import org.mapstruct.Mapping;
 import org.springframework.data.domain.Page;
 import org.springframework.data.domain.PageImpl;
 
@@ -15,6 +16,14 @@
 @AnnotateWith(NullUnmarked.class)
 @NullMarked
 public interface EmailMapper {
+    @Mapping(source = "content.subject", target = "subject")
+    @Mapping(source = "content.fromAddress", target = "fromAddress")
+    @Mapping(source = "content.fromName", target = "fromName")
+    @Mapping(source = "content.replyToAddress", target = "replyToAddress")
+    @Mapping(source = "content.replyToName", target = "replyToName")
+    @Mapping(source = "content.recipients", target = "recipients")
+    @Mapping(source = "content.textBody", target = "textBody")
+    @Mapping(source = "content.htmlBody", target = "htmlBody")
     EmailDto toDto(Email model);
 
     List<EmailDto> toDto(List<Email> model);

src/main/java/it/aboutbits/springboot/emailservice/lib/application/EmailServiceMigrator.java

@@ -69,6 +69,15 @@ updated_at           timestamp with time zone default now()           not null,
 
                  alter table email_service_emails add column if not exists reply_to_address text;
                  alter table email_service_emails add column if not exists reply_to_name text;
+
+                 alter table email_service_emails add column if not exists attempts int default 0 not null;
+                 alter table email_service_emails add column if not exists execution_start_time timestamp with time zone;
+                 alter table email_service_emails add column if not exists execution_end_time timestamp with time zone;
+
+                 update email_service_emails
+                    set execution_end_time = sent_at
+                  where sent_at is not null
+                    and execution_end_time is null;
                 """
                 //@formatter:on
         );

src/main/java/it/aboutbits/springboot/emailservice/lib/application/ManageEmail.java

@@ -9,21 +9,27 @@
 import it.aboutbits.springboot.emailservice.lib.jpa.EmailRepository;
 import it.aboutbits.springboot.emailservice.lib.model.Email;
 import it.aboutbits.springboot.emailservice.lib.model.EmailAttachment;
+import it.aboutbits.springboot.emailservice.lib.model.EmailContent;
 import jakarta.mail.MessagingException;
 import jakarta.validation.Valid;
 import lombok.extern.slf4j.Slf4j;
 import org.jspecify.annotations.NullMarked;
 import org.jspecify.annotations.Nullable;
 import org.springframework.core.io.ByteArrayResource;
-import org.springframework.mail.MailException;
 import org.springframework.mail.javamail.JavaMailSender;
 import org.springframework.mail.javamail.MimeMessageHelper;
+import org.springframework.transaction.PlatformTransactionManager;
+import org.springframework.transaction.TransactionDefinition;
+import org.springframework.transaction.annotation.Transactional;
+import org.springframework.transaction.support.TransactionTemplate;
 import org.springframework.validation.annotation.Validated;
 
 import java.io.IOException;
+import java.time.Duration;
 import java.time.OffsetDateTime;
 import java.util.HashSet;
 import java.util.List;
+import java.util.Optional;
 import java.util.Set;
 
 
@@ -35,17 +41,28 @@ public class ManageEmail {
     private final JavaMailSender mailSender;
     private final AttachmentDataSource attachmentDataSource;
     private final EmailMapper emailMapper;
+    private final int maxAttempts;
+    private final Duration schedulerInterval;
+    private final TransactionTemplate transactionTemplate;
 
     public ManageEmail(
             EmailRepository emailRepository,
             JavaMailSender mailSender,
             AttachmentDataSource attachmentDataSource,
-            final EmailMapper emailMapper
+            EmailMapper emailMapper,
+            int maxAttempts,
+            Duration schedulerInterval,
+            PlatformTransactionManager transactionManager
     ) {
         this.emailRepository = emailRepository;
         this.mailSender = mailSender;
         this.attachmentDataSource = attachmentDataSource;
         this.emailMapper = emailMapper;
+        this.maxAttempts = maxAttempts;
+        this.schedulerInterval = schedulerInterval;
+        // Persist the final email state in its own, independent transaction for sendOrFail to never make it roll back
+        this.transactionTemplate = new TransactionTemplate(transactionManager);
+        this.transactionTemplate.setPropagationBehavior(TransactionDefinition.PROPAGATION_REQUIRES_NEW);
     }
 
     public EmailDto schedule(@Valid EmailParameter parameter) throws EmailException {
@@ -69,33 +86,60 @@ public EmailDto sendOrFail(@Valid EmailParameter parameter) throws EmailExceptio
             throw new EmailException(e);
         }
 
-        var savedEmail = send(email);
+        email.setExecutionStartTime(OffsetDateTime.now());
+        email.incrementAttempts();
 
-        if (savedEmail.hasFailed()) {
+        try {
+            sendMail(email);
+            email.setState(EmailState.SENT);
+            email.setExecutionEndTime(OffsetDateTime.now());
+        } catch (MessagingException | AttachmentException | IOException | RuntimeException e) {
+            log.error("Failed to send email: {}", email.getId(), e);
+            email.setState(EmailState.ERROR);
+            email.setExecutionEndTime(OffsetDateTime.now());
+            email.setErrorMessage(e.getMessage());
+        }
+
+        var savedEmail = transactionTemplate.execute(_ -> emailRepository.save(email));
+
+        if (savedEmail.getState() == EmailState.ERROR) {
             throw new EmailException("Failed to send email [id=%s, providerMessage=%s]"
                                              .formatted(savedEmail.getId(), savedEmail.getErrorMessage()));
         }
 
         return emailMapper.toDto(savedEmail);
     }
 
-    Email send(Email email) {
-        if (EmailState.SENT.equals(email.getState())) {
-            return email;
-        }
+    // Atomically transitions the row from PENDING/stale-SENDING into SENDING
+    @Transactional
+    Optional<Email> tryClaimForSend(long id, OffsetDateTime staleSendingBefore) {
+        var claimed = emailRepository.claimForSend(id, OffsetDateTime.now(), staleSendingBefore);
+        return claimed == 0 ? Optional.empty() : emailRepository.findById(id);
+    }
 
+    // Actually try sending the claimed email
+    Email completeClaimedSend(Email email) {
         try {
             sendMail(email);
             email.setState(EmailState.SENT);
-            email.setErrorMessage("");
-            email.setSentAt(OffsetDateTime.now());
-        } catch (MailException | MessagingException | AttachmentException | IOException e) {
-            log.error("Failed to send email: " + email.getId(), e);
+            email.setExecutionEndTime(OffsetDateTime.now());
+            email.setErrorMessage(null);
+        } catch (Exception e) {
+            email.setExecutionEndTime(OffsetDateTime.now());
             email.setErrorMessage(e.getMessage());
-            email.setState(EmailState.ERROR);
+            if (email.getAttempts() >= maxAttempts) {
+                log.error("Failed to send email: {}", email.getId(), e);
+                email.setState(EmailState.ERROR);
+            } else {
+                log.warn("Failed to send email: {}; Will be tried again", email.getId(), e);
+                email.setState(EmailState.PENDING);
+                email.setScheduledAt(
+                        OffsetDateTime.now()
+                                .plus(schedulerInterval.multipliedBy((long) Math.pow(2, email.getAttempts())))
+                );
+            }
         }
-
-        return emailRepository.save(email);
+        return transactionTemplate.execute(_ -> emailRepository.save(email));
     }
 
     void cleanupAttachments(final Email email) throws AttachmentException {
@@ -106,49 +150,17 @@ void cleanupAttachments(final Email email) throws AttachmentException {
         emailRepository.save(email);
     }
 
-    private Email fromParameter(EmailParameter parameter) throws AttachmentException {
-        var emailData = parameter.email();
-
-        final var email = new Email();
-        email.setState(EmailState.PENDING);
-        email.setScheduledAt(parameter.scheduledAt());
-        email.setSubject(emailData.subject());
-        email.setTextBody(emailData.textBody());
-        email.setHtmlBody(emailData.htmlBody());
-        email.setRecipients(emailData.recipients());
-        email.setFromAddress(emailData.fromAddress());
-        email.setFromName(emailData.fromName());
-        email.setReplyToAddress(emailData.replyToAddress());
-        email.setReplyToName(emailData.replyToName());
-
-        var attachments = new HashSet<EmailAttachment>();
-        for (var attachment : parameter.email().attachments()) {
-            var reference = attachmentDataSource.storeAttachmentPayload(attachment.payload());
-
-            var emailAttachment = new EmailAttachment();
-            emailAttachment.setEmail(email);
-            emailAttachment.setContentType(attachment.contentType());
-            emailAttachment.setFileName(attachment.fileName());
-            emailAttachment.setFileReference(reference);
-
-            attachments.add(emailAttachment);
-        }
-
-        email.setAttachments(attachments);
-
-        return email;
-    }
-
     private void sendMail(Email email) throws MessagingException, IOException, AttachmentException {
+        var content = email.getContent();
         sendMail(
-                email.getFromAddress(),
-                email.getFromName(),
-                email.getReplyToAddress(),
-                email.getReplyToName(),
-                email.getRecipients(),
-                email.getSubject(),
-                email.getHtmlBody(),
-                email.getTextBody(),
+                content.fromAddress(),
+                content.fromName(),
+                content.replyToAddress(),
+                content.replyToName(),
+                content.recipients(),
+                content.subject(),
+                content.htmlBody(),
+                content.textBody(),
                 email.getAttachments()
         );
     }
@@ -194,7 +206,41 @@ private void sendMail(
             payload.close();
         }
 
-
         mailSender.send(message);
     }
+
+    private Email fromParameter(EmailParameter parameter) throws AttachmentException {
+        var emailData = parameter.email();
+
+        var email = new Email();
+        email.setState(EmailState.PENDING);
+        email.setScheduledAt(parameter.scheduledAt());
+        email.setContent(new EmailContent(
+                emailData.subject(),
+                emailData.fromAddress(),
+                emailData.fromName(),
+                emailData.replyToAddress(),
+                emailData.replyToName(),
+                emailData.recipients(),
+                emailData.textBody(),
+                emailData.htmlBody()
+        ));
+
+        var attachments = new HashSet<EmailAttachment>();
+        for (var attachment : parameter.email().attachments()) {
+            var reference = attachmentDataSource.storeAttachmentPayload(attachment.payload());
+
+            var emailAttachment = new EmailAttachment();
+            emailAttachment.setEmail(email);
+            emailAttachment.setContentType(attachment.contentType());
+            emailAttachment.setFileName(attachment.fileName());
+            emailAttachment.setFileReference(reference);
+
+            attachments.add(emailAttachment);
+        }
+
+        email.setAttachments(attachments);
+
+        return email;
+    }
 }