diff --git a/readme.md b/readme.md index 19c5434..cf86fff 100644 --- a/readme.md +++ b/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: diff --git a/src/main/java/it/aboutbits/springboot/emailservice/EmailServiceConfiguration.java b/src/main/java/it/aboutbits/springboot/emailservice/EmailServiceConfiguration.java index 535054e..c882776 100644 --- a/src/main/java/it/aboutbits/springboot/emailservice/EmailServiceConfiguration.java +++ b/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 callbacks) { - return new SendScheduledEmails(queryEmail, manageEmail, callbacks); + public SendScheduledEmails sendScheduledEmails( + QueryEmail queryEmail, + ManageEmail manageEmail, + List 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 callbacks) { + public CleanupAttachmentFiles cleanupAttachments( + QueryEmail queryEmail, + ManageEmail manageEmail, + List callbacks + ) { return new CleanupAttachmentFiles(queryEmail, manageEmail, callbacks); } diff --git a/src/main/java/it/aboutbits/springboot/emailservice/lib/EmailDto.java b/src/main/java/it/aboutbits/springboot/emailservice/lib/EmailDto.java index 11ff3ac..eefe73c 100644 --- a/src/main/java/it/aboutbits/springboot/emailservice/lib/EmailDto.java +++ b/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, diff --git a/src/main/java/it/aboutbits/springboot/emailservice/lib/EmailState.java b/src/main/java/it/aboutbits/springboot/emailservice/lib/EmailState.java index 4d2ed5d..5298f69 100644 --- a/src/main/java/it/aboutbits/springboot/emailservice/lib/EmailState.java +++ b/src/main/java/it/aboutbits/springboot/emailservice/lib/EmailState.java @@ -5,6 +5,7 @@ @NullMarked public enum EmailState { PENDING, + SENDING, SENT, ERROR } diff --git a/src/main/java/it/aboutbits/springboot/emailservice/lib/application/EmailMapper.java b/src/main/java/it/aboutbits/springboot/emailservice/lib/application/EmailMapper.java index 48aa8c9..ecf528c 100644 --- a/src/main/java/it/aboutbits/springboot/emailservice/lib/application/EmailMapper.java +++ b/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 toDto(List model); diff --git a/src/main/java/it/aboutbits/springboot/emailservice/lib/application/EmailServiceMigrator.java b/src/main/java/it/aboutbits/springboot/emailservice/lib/application/EmailServiceMigrator.java index 617732a..919411a 100644 --- a/src/main/java/it/aboutbits/springboot/emailservice/lib/application/EmailServiceMigrator.java +++ b/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 ); diff --git a/src/main/java/it/aboutbits/springboot/emailservice/lib/application/ManageEmail.java b/src/main/java/it/aboutbits/springboot/emailservice/lib/application/ManageEmail.java index 3346aac..7563888 100644 --- a/src/main/java/it/aboutbits/springboot/emailservice/lib/application/ManageEmail.java +++ b/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,9 +86,23 @@ 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())); } @@ -79,23 +110,36 @@ public EmailDto sendOrFail(@Valid EmailParameter parameter) throws EmailExceptio 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 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(); - 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(); + 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; + } } diff --git a/src/main/java/it/aboutbits/springboot/emailservice/lib/application/QueryEmail.java b/src/main/java/it/aboutbits/springboot/emailservice/lib/application/QueryEmail.java index 7c43588..36da5bf 100644 --- a/src/main/java/it/aboutbits/springboot/emailservice/lib/application/QueryEmail.java +++ b/src/main/java/it/aboutbits/springboot/emailservice/lib/application/QueryEmail.java @@ -5,7 +5,6 @@ import it.aboutbits.springboot.emailservice.lib.EmailState; import it.aboutbits.springboot.emailservice.lib.jpa.EmailRepository; import it.aboutbits.springboot.emailservice.lib.model.Email; -import jakarta.persistence.EntityManager; import lombok.RequiredArgsConstructor; import org.jspecify.annotations.NullMarked; import org.springframework.data.domain.Page; @@ -23,7 +22,6 @@ public class QueryEmail { private final EmailRepository emailRepository; private final EmailMapper emailMapper; - private final EntityManager entityManager; public Page paginatedByState(EmailState state, PageRequest pageParameter) { var pageRequest = PageRequest.of( @@ -42,30 +40,15 @@ public List byIds(Collection ids) { return emailMapper.toDto(emailRepository.findByIdIn(ids)); } - List readyToSend() { - var entityGraph = entityManager.getEntityGraph("email_service_emails-entity-graph"); - return entityManager.createQuery( - """ - SELECT e from Email e WHERE e.scheduledAt < :scheduledBefore AND e.state IN ( - it.aboutbits.springboot.emailservice.lib.EmailState.PENDING, - it.aboutbits.springboot.emailservice.lib.EmailState.ERROR - ) - """, Email.class - ) - .setParameter("scheduledBefore", OffsetDateTime.now()) - .setHint("jakarta.persistence.fetchgraph", entityGraph) - .getResultList(); + List candidateIdsToSend(OffsetDateTime staleSendingBefore) { + return emailRepository.findCandidateIdsToSend( + OffsetDateTime.now(), + staleSendingBefore + ); } List readyToCleanup() { - var entityGraph = entityManager.getEntityGraph("email_service_emails-entity-graph"); - return entityManager.createQuery( - """ - SELECT e from Email e WHERE e.attachmentsCleaned=false AND e.state=it.aboutbits.springboot.emailservice.lib.EmailState.SENT - """, Email.class - ) - .setHint("jakarta.persistence.fetchgraph", entityGraph) - .getResultList(); + return emailRepository.findReadyToCleanup(); } public Optional byId(long id) { diff --git a/src/main/java/it/aboutbits/springboot/emailservice/lib/application/SendScheduledEmails.java b/src/main/java/it/aboutbits/springboot/emailservice/lib/application/SendScheduledEmails.java index f1789d0..4926336 100644 --- a/src/main/java/it/aboutbits/springboot/emailservice/lib/application/SendScheduledEmails.java +++ b/src/main/java/it/aboutbits/springboot/emailservice/lib/application/SendScheduledEmails.java @@ -2,15 +2,16 @@ import it.aboutbits.springboot.emailservice.lib.EmailSchedulerCallback; -import lombok.RequiredArgsConstructor; +import it.aboutbits.springboot.emailservice.lib.model.Email; import lombok.extern.log4j.Log4j2; import org.jspecify.annotations.NullMarked; import org.springframework.scheduling.annotation.Scheduled; import java.time.Duration; +import java.time.OffsetDateTime; import java.util.List; +import java.util.Optional; -@RequiredArgsConstructor @Log4j2 @NullMarked public class SendScheduledEmails { @@ -19,27 +20,67 @@ public class SendScheduledEmails { private final QueryEmail queryEmail; private final ManageEmail manageEmail; private final List callbacks; + private final Duration stuckSendingRecoveryThreshold; private long lastInfoLogMillis = System.currentTimeMillis(); private long silentRuns = 0; private boolean firstRun = true; + public SendScheduledEmails( + QueryEmail queryEmail, + ManageEmail manageEmail, + List callbacks, + Duration stuckSendingRecoveryThreshold + ) { + this.queryEmail = queryEmail; + this.manageEmail = manageEmail; + this.callbacks = callbacks; + this.stuckSendingRecoveryThreshold = stuckSendingRecoveryThreshold; + } + @Scheduled(initialDelayString = "${aboutbits.emailservice.scheduling.interval:30000}", fixedDelayString = "${aboutbits.emailservice.scheduling.interval:30000}") void sendEmails() { logStartOfPass(); - var emailsToSend = queryEmail.readyToSend(); + var staleSendingBefore = OffsetDateTime.now().minus(stuckSendingRecoveryThreshold); + var candidateIds = queryEmail.candidateIdsToSend(staleSendingBefore); + var countClaimed = 0; var countSent = 0; var countError = 0; - for (var email : emailsToSend) { - var updatedEmail = manageEmail.send(email); - switch (updatedEmail.getState()) { - case ERROR -> countError++; - case SENT -> countSent++; - default -> log.warn( - JOB_DESCRIPTION + " | Job produced an invalid notification result state: {}.", - updatedEmail.getState().name() + for (var id : candidateIds) { + Optional claimed; + claimed = manageEmail.tryClaimForSend(id, staleSendingBefore); + + if (claimed.isEmpty()) { + // Lost race to another pod; Skip + continue; + } + countClaimed++; + + try { + var updated = manageEmail.completeClaimedSend(claimed.get()); + switch (updated.getState()) { + // We do not count PENDING emails as errors since they will + // be retried and eventually end up in one of the two buckets. + case ERROR -> countError++; + case SENT -> countSent++; + default -> log.warn( + JOB_DESCRIPTION + " | Job produced an invalid notification result state: {}.", + updated.getState().name() + ); + } + } catch (Exception e) { + // Best-effort visibility only: we cannot reliably recover here. The email may already have been + // sent but persisting the result failed, so the row stays in SENDING and will be re-claimed and + // re-sent after the recovery threshold (possible duplicate delivery). + countError++; + log.error( + JOB_DESCRIPTION + " | Failed to complete send for email: {}. It may have been sent already; " + + "if the result could not be persisted the row stays in SENDING and will be re-sent " + + "after the recovery threshold (possible duplicate delivery).", + id, + e ); } } @@ -48,7 +89,7 @@ void sendEmails() { for (var callback : callbacks) { callback.report(new EmailSchedulerCallback.Report( - emailsToSend.size(), + countClaimed, countSent, countError )); diff --git a/src/main/java/it/aboutbits/springboot/emailservice/lib/jpa/EmailRepository.java b/src/main/java/it/aboutbits/springboot/emailservice/lib/jpa/EmailRepository.java index ba3100c..4c8c150 100644 --- a/src/main/java/it/aboutbits/springboot/emailservice/lib/jpa/EmailRepository.java +++ b/src/main/java/it/aboutbits/springboot/emailservice/lib/jpa/EmailRepository.java @@ -8,7 +8,11 @@ import org.springframework.data.domain.PageRequest; import org.springframework.data.jpa.repository.EntityGraph; import org.springframework.data.jpa.repository.JpaRepository; +import org.springframework.data.jpa.repository.Modifying; +import org.springframework.data.jpa.repository.Query; +import org.springframework.data.repository.query.Param; +import java.time.OffsetDateTime; import java.util.Collection; import java.util.List; import java.util.Optional; @@ -32,4 +36,59 @@ public interface EmailRepository extends JpaRepository { @EntityGraph(value = Email.DEFAULT_ENTITY_GRAPH) List findByIdIn(Collection ids); + + // Plain read, no locking -> two pods may see overlapping candidate sets. + // The atomic UPDATE in claimForSend arbitrates the actual claim. + // Includes SENDING rows abandoned by crashed pods (past the stale threshold). + @Query(""" + select e.id from Email e + where e.scheduledAt < :now + and ( + e.state = it.aboutbits.springboot.emailservice.lib.EmailState.PENDING + or ( + e.state = it.aboutbits.springboot.emailservice.lib.EmailState.SENDING + and e.executionStartTime < :staleSendingBefore + ) + ) + order by e.scheduledAt + """) + List findCandidateIdsToSend( + @Param("now") OffsetDateTime now, + @Param("staleSendingBefore") OffsetDateTime staleSendingBefore + ); + + // Atomic compare-and-set claim: transitions a single row into SENDING if its + // current state is claimable (PENDING, or SENDING abandoned by a crashed pod past the stale threshold). + // Concurrent updates are serialized against the same row, so EXACTLY ONE caller gets returned 1. + @Modifying(clearAutomatically = true, flushAutomatically = true) + @Query(""" + update Email e + set e.state = it.aboutbits.springboot.emailservice.lib.EmailState.SENDING, + e.executionStartTime = :now, + e.executionEndTime = null, + e.errorMessage = null, + e.attempts = e.attempts + 1 + where e.id = :id + and e.scheduledAt < :now + and ( + e.state = it.aboutbits.springboot.emailservice.lib.EmailState.PENDING + or ( + e.state = it.aboutbits.springboot.emailservice.lib.EmailState.SENDING + and e.executionStartTime < :staleSendingBefore + ) + ) + """) + int claimForSend( + @Param("id") long id, + @Param("now") OffsetDateTime now, + @Param("staleSendingBefore") OffsetDateTime staleSendingBefore + ); + + @EntityGraph(value = Email.DEFAULT_ENTITY_GRAPH) + @Query(""" + select e from Email e + where e.attachmentsCleaned = false + and e.state = it.aboutbits.springboot.emailservice.lib.EmailState.SENT + """) + List findReadyToCleanup(); } diff --git a/src/main/java/it/aboutbits/springboot/emailservice/lib/model/Email.java b/src/main/java/it/aboutbits/springboot/emailservice/lib/model/Email.java index 88a990e..2c09720 100644 --- a/src/main/java/it/aboutbits/springboot/emailservice/lib/model/Email.java +++ b/src/main/java/it/aboutbits/springboot/emailservice/lib/model/Email.java @@ -2,6 +2,7 @@ import it.aboutbits.springboot.emailservice.lib.EmailState; import jakarta.persistence.CascadeType; +import jakarta.persistence.Embedded; import jakarta.persistence.Entity; import jakarta.persistence.EnumType; import jakarta.persistence.Enumerated; @@ -18,24 +19,16 @@ import lombok.NoArgsConstructor; import lombok.Setter; import org.hibernate.annotations.CreationTimestamp; -import org.hibernate.annotations.JdbcTypeCode; import org.hibernate.annotations.UpdateTimestamp; -import org.hibernate.type.SqlTypes; import org.jspecify.annotations.NullUnmarked; import org.jspecify.annotations.Nullable; import java.time.OffsetDateTime; -import java.util.List; +import java.util.HashSet; import java.util.Set; import static it.aboutbits.springboot.emailservice.lib.model.Email.DEFAULT_ENTITY_GRAPH; -@NamedEntityGraph( - name = "email_service_emails-entity-graph", - attributeNodes = { - @NamedAttributeNode("attachments") - } -) @Entity @Getter @Setter @@ -55,34 +48,24 @@ public class Email { @Enumerated(EnumType.STRING) private EmailState state; - private String subject; - - private String fromAddress; - private String fromName; - - @Nullable - private String replyToAddress; - @Nullable - private String replyToName; - - @JdbcTypeCode(SqlTypes.JSON) - private List recipients; - - private String textBody; - private String htmlBody; + @Embedded + private EmailContent content; + @Builder.Default @OneToMany(cascade = CascadeType.PERSIST, mappedBy = "email", orphanRemoval = true) - private Set attachments; + private Set attachments = new HashSet<>(); @Builder.Default private boolean attachmentsCleaned = false; private OffsetDateTime scheduledAt; @Nullable - private OffsetDateTime sentAt; - + private OffsetDateTime executionStartTime; @Nullable - private OffsetDateTime errorAt; + private OffsetDateTime executionEndTime; + + private int attempts = 0; + @Nullable private String errorMessage; @@ -92,11 +75,7 @@ public class Email { @UpdateTimestamp private OffsetDateTime updatedAt; - public boolean isSent() { - return EmailState.SENT.equals(state); - } - - public boolean hasFailed() { - return EmailState.ERROR.equals(state); + public void incrementAttempts() { + this.attempts++; } } diff --git a/src/main/java/it/aboutbits/springboot/emailservice/lib/model/EmailContent.java b/src/main/java/it/aboutbits/springboot/emailservice/lib/model/EmailContent.java new file mode 100644 index 0000000..dc68116 --- /dev/null +++ b/src/main/java/it/aboutbits/springboot/emailservice/lib/model/EmailContent.java @@ -0,0 +1,30 @@ +package it.aboutbits.springboot.emailservice.lib.model; + +import jakarta.persistence.Embeddable; +import org.hibernate.annotations.JdbcTypeCode; +import org.hibernate.type.SqlTypes; +import org.jspecify.annotations.NullMarked; +import org.jspecify.annotations.Nullable; + +import java.util.List; + +@Embeddable +@NullMarked +public record EmailContent( + String subject, + + String fromAddress, + String fromName, + + @Nullable + String replyToAddress, + @Nullable + String replyToName, + + @JdbcTypeCode(SqlTypes.JSON) + List recipients, + + String textBody, + String htmlBody +) { +} diff --git a/src/main/resources/META-INF/additional-spring-configuration-metadata.json b/src/main/resources/META-INF/additional-spring-configuration-metadata.json index 6653b27..7b42404 100644 --- a/src/main/resources/META-INF/additional-spring-configuration-metadata.json +++ b/src/main/resources/META-INF/additional-spring-configuration-metadata.json @@ -23,6 +23,18 @@ "type": "java.lang.Long", "description": "Specifies the milliseconds delay between runs of the scheduler.", "defaultValue": 30000 + }, + { + "name": "aboutbits.emailservice.scheduling.stuck-sending-recovery-threshold", + "type": "java.time.Duration", + "description": "How long an email may stay in the SENDING state before being considered abandoned (crashed pod) and eligible to be re-claimed by another pod. 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.", + "defaultValue": "PT30M" + }, + { + "name": "aboutbits.emailservice.scheduling.max-attempts", + "type": "java.lang.Integer", + "description": "Maximum number of send attempts before an email is marked as ERROR. Applies to the scheduled retry loop; the synchronous sendOrFail path is fail-fast and does not retry. Failed attempts are retried with exponential backoff (scheduling.interval x 2^attempts); with the defaults a persistently failing email runs attempt 1 -> +60s -> attempt 2 -> +120s -> attempt 3 -> ERROR.", + "defaultValue": 3 } ] } diff --git a/src/test/java/it/aboutbits/springboot/emailservice/lib/application/SendScheduledEmailsTest.java b/src/test/java/it/aboutbits/springboot/emailservice/lib/application/SendScheduledEmailsTest.java new file mode 100644 index 0000000..1af6242 --- /dev/null +++ b/src/test/java/it/aboutbits/springboot/emailservice/lib/application/SendScheduledEmailsTest.java @@ -0,0 +1,221 @@ +package it.aboutbits.springboot.emailservice.lib.application; + +import it.aboutbits.springboot.emailservice.lib.EmailState; +import it.aboutbits.springboot.emailservice.lib.jpa.EmailRepository; +import it.aboutbits.springboot.emailservice.support.database.WithPostgres; +import it.aboutbits.springboot.emailservice.support.database.factory.EmailFactory; +import jakarta.mail.internet.MimeMessage; +import org.jspecify.annotations.NullMarked; +import org.junit.jupiter.api.BeforeEach; +import org.junit.jupiter.api.Test; +import org.springframework.beans.factory.annotation.Autowired; +import org.springframework.boot.test.context.SpringBootTest; +import org.springframework.mail.MailSendException; +import org.springframework.mail.javamail.JavaMailSender; +import org.springframework.test.context.bean.override.mockito.MockitoSpyBean; + +import java.time.OffsetDateTime; +import java.time.temporal.ChronoUnit; +import java.util.concurrent.CountDownLatch; +import java.util.concurrent.Executors; +import java.util.concurrent.TimeUnit; +import java.util.stream.IntStream; + +import static org.assertj.core.api.Assertions.assertThat; +import static org.mockito.ArgumentMatchers.any; +import static org.mockito.Mockito.doNothing; +import static org.mockito.Mockito.doThrow; +import static org.mockito.Mockito.times; +import static org.mockito.Mockito.verify; + +@SpringBootTest(properties = { + "aboutbits.emailservice.scheduling.max-attempts=3", + "aboutbits.emailservice.scheduling.stuck-sending-recovery-threshold=PT5M", + "aboutbits.emailservice.scheduling.interval=30000" +}) +@WithPostgres +@NullMarked +class SendScheduledEmailsTest { + private static final int MAX_ATTEMPTS = 3; + private static final int SCHEDULER_INTERVAL_SECONDS = 30; + + @MockitoSpyBean + JavaMailSender javaMailSender; + + @Autowired + EmailRepository emailRepository; + + @Autowired + SendScheduledEmails sendScheduledEmails; + + @BeforeEach + void setup() { + doNothing().when(javaMailSender).send(any(MimeMessage.class)); + } + + @Test + void givenPendingEmails_sendEmails_shouldMarkAllAsSent() { + emailRepository.saveAll(IntStream.range(0, 3).mapToObj(_ -> EmailFactory.once().build()).toList()); + + sendScheduledEmails.sendEmails(); + + assertThat(emailRepository.findAll()) + .hasSize(3) + .allMatch(email -> email.getState() == EmailState.SENT) + .allMatch(email -> email.getExecutionStartTime() != null) + .allMatch(email -> email.getExecutionEndTime() != null) + .allMatch(email -> email.getAttempts() == 1); + verify(javaMailSender, times(3)).send(any(MimeMessage.class)); + } + + @Test + void givenRetryableFailure_sendEmails_shouldRescheduleWithBackoffAndIncrementAttempts() { + doThrow(new MailSendException("smtp blip")).when(javaMailSender).send(any(MimeMessage.class)); + emailRepository.save(EmailFactory.once().build()); + + var beforePass = OffsetDateTime.now(); + sendScheduledEmails.sendEmails(); + + assertThat(emailRepository.findAll()) + .singleElement() + .satisfies(email -> { + assertThat(email.getState()).isEqualTo(EmailState.PENDING); + assertThat(email.getAttempts()).isEqualTo(1); + // First retry backoff: 2^attempts * scheduler interval = 2 * scheduler interval + assertThat(email.getScheduledAt()) + .isAfterOrEqualTo(beforePass.plusSeconds(2L * SCHEDULER_INTERVAL_SECONDS)); + assertThat(email.getErrorMessage()).contains("smtp blip"); + assertThat(email.getExecutionEndTime()).isNotNull(); + }); + } + + @Test + void givenAttemptsAlreadyAtBudget_sendEmails_shouldEscalateToError() { + doThrow(new MailSendException("smtp down")).when(javaMailSender).send(any(MimeMessage.class)); + emailRepository.save( + EmailFactory.once() + .attempts(MAX_ATTEMPTS - 1) + .scheduledAt(OffsetDateTime.now().minusSeconds(30)) + .build() + ); + + sendScheduledEmails.sendEmails(); + + assertThat(emailRepository.findAll()) + .singleElement() + .satisfies(email -> { + // The atomic claim UPDATE incremented attempts from MAX_ATTEMPTS-1 to MAX_ATTEMPTS, + // which is equal to the threshold, so the failure escalates to ERROR. + assertThat(email.getState()).isEqualTo(EmailState.ERROR); + assertThat(email.getAttempts()).isEqualTo(MAX_ATTEMPTS); + assertThat(email.getErrorMessage()).contains("smtp down"); + }); + } + + @Test + void givenErrorRow_sendEmails_shouldNotRetry() { + emailRepository.save( + EmailFactory.once() + .state(EmailState.ERROR) + .attempts(MAX_ATTEMPTS) + .scheduledAt(OffsetDateTime.now().minusMinutes(1)) + .errorMessage("previous permanent failure") + .build() + ); + + sendScheduledEmails.sendEmails(); + + assertThat(emailRepository.findAll()) + .singleElement() + .satisfies(email -> { + assertThat(email.getState()).isEqualTo(EmailState.ERROR); + assertThat(email.getAttempts()).isEqualTo(MAX_ATTEMPTS); + }); + verify(javaMailSender, times(0)).send(any(MimeMessage.class)); + } + + @Test + void givenStuckSendingRow_sendEmails_shouldRecoverAndResend() { + var staleStart = OffsetDateTime.now().minusMinutes(10); + emailRepository.save( + EmailFactory.once() + .state(EmailState.SENDING) + .attempts(1) + .executionStartTime(staleStart) + .scheduledAt(OffsetDateTime.now().minusSeconds(60)) + .build() + ); + + sendScheduledEmails.sendEmails(); + + assertThat(emailRepository.findAll()) + .singleElement() + .satisfies(email -> { + assertThat(email.getState()).isEqualTo(EmailState.SENT); + assertThat(email.getAttempts()).isEqualTo(2); + assertThat(email.getExecutionStartTime()).isAfter(staleStart); + }); + verify(javaMailSender, times(1)).send(any(MimeMessage.class)); + } + + @Test + void givenFreshSendingRowWithinThreshold_sendEmails_shouldNotStealFromOtherPod() { + // executionStartTime within the 5-minute threshold means another pod is legitimately + // sending this right now; we must not re-claim it. + var recentStart = OffsetDateTime.now().minusSeconds(30).truncatedTo(ChronoUnit.MICROS); + emailRepository.save( + EmailFactory.once() + .state(EmailState.SENDING) + .attempts(1) + .executionStartTime(recentStart) + .scheduledAt(OffsetDateTime.now().minusSeconds(60)) + .build() + ); + + sendScheduledEmails.sendEmails(); + + assertThat(emailRepository.findAll()) + .singleElement() + .satisfies(email -> { + assertThat(email.getState()).isEqualTo(EmailState.SENDING); + assertThat(email.getAttempts()).isEqualTo(1); + assertThat(email.getExecutionStartTime()).isEqualTo(recentStart); + }); + verify(javaMailSender, times(0)).send(any(MimeMessage.class)); + } + + @Test + void givenConcurrentPasses_sendEmails_shouldSendEachEmailExactlyOnce() throws Exception { + var totalEmails = 25; + var workerThreads = 4; + var passesPerWorker = 10; + emailRepository.saveAll(IntStream.range(0, totalEmails).mapToObj(_ -> EmailFactory.once().build()).toList()); + + var startGate = new CountDownLatch(1); + var executor = Executors.newFixedThreadPool(workerThreads); + try { + for (var t = 0; t < workerThreads; t++) { + executor.submit(() -> { + startGate.await(); + for (var i = 0; i < passesPerWorker; i++) { + sendScheduledEmails.sendEmails(); + } + return null; + }); + } + startGate.countDown(); + executor.shutdown(); + assertThat(executor.awaitTermination(60, TimeUnit.SECONDS)).isTrue(); + } finally { + if (!executor.isTerminated()) { + executor.shutdownNow(); + } + } + + assertThat(emailRepository.findAll()) + .hasSize(totalEmails) + .allMatch(email -> email.getState() == EmailState.SENT) + .allMatch(email -> email.getAttempts() == 1); + verify(javaMailSender, times(totalEmails)).send(any(MimeMessage.class)); + } +} diff --git a/src/test/java/it/aboutbits/springboot/emailservice/support/database/factory/EmailFactory.java b/src/test/java/it/aboutbits/springboot/emailservice/support/database/factory/EmailFactory.java index 1c37b19..e8632cf 100644 --- a/src/test/java/it/aboutbits/springboot/emailservice/support/database/factory/EmailFactory.java +++ b/src/test/java/it/aboutbits/springboot/emailservice/support/database/factory/EmailFactory.java @@ -2,6 +2,7 @@ import it.aboutbits.springboot.emailservice.lib.EmailState; import it.aboutbits.springboot.emailservice.lib.model.Email; +import it.aboutbits.springboot.emailservice.lib.model.EmailContent; import it.aboutbits.springboot.testing.testdata.FakerExtended; import org.jspecify.annotations.NullMarked; @@ -20,14 +21,16 @@ public static Email.EmailBuilder once() { return Email.builder() .state(EmailState.PENDING) - .subject("Email subject") - .textBody(body) - .htmlBody("

" + body + "

") .scheduledAt(OffsetDateTime.now()) - .fromAddress(FAKER.internet().emailAddress()) - .fromName(FAKER.name().fullName()) - .replyToAddress(FAKER.internet().emailAddress()) - .replyToName(FAKER.name().fullName()) - .recipients(List.of(FAKER.internet().emailAddress(), FAKER.internet().emailAddress())); + .content(new EmailContent( + "Email subject", + FAKER.internet().emailAddress(), + FAKER.name().fullName(), + FAKER.internet().emailAddress(), + FAKER.name().fullName(), + List.of(FAKER.internet().emailAddress(), FAKER.internet().emailAddress()), + body, + "

" + body + "

" + )); } }