requested changes

Author: J0nasMayr

readme.md

@@ -80,34 +80,13 @@ The following configuration options are available:
 | `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`    | PT5M    | How long an email may stay in `SENDING` before being considered abandoned (crashed pod) and eligible to be re-claimed. |
-| `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.        |
+| `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. Each pass performs
-three independent steps:
-
-1. **Candidate scan** — a plain `SELECT` returns up to `batch-size` ids of
-   emails whose `scheduled_at` has passed and whose state is `PENDING` or `SENDING`
-   older than `stuck-sending-recovery-threshold` (crashed-pod recovery). `ERROR`
-   is terminal: rows that exhausted `max-attempts` are never re-picked
-   automatically — an operator can reset them to `PENDING` if a retry is desired.
-   Two pods may see overlapping ids at this step, and that's fine — the atomic
-   claim below arbitrates.
-2. **Atomic claim** — for each candidate id, a single compare-and-set
-   `UPDATE … SET state = SENDING … WHERE id = ? AND state IN (PENDING, ERROR, staleSENDING)`
-   is issued. The database serializes concurrent updates against the same row so
-   exactly one pod sees `rowsAffected = 1` and owns that email; the other pods
-   see `0` and move on.
-3. **Send and persist** — the winning pod calls SMTP outside any database
-   transaction and then writes the final `SENT` / `ERROR` / `PENDING` state through a small `save()`.
-
-Crash recovery: if a pod dies between "claim" and "persist result", the row
-stays in `SENDING` with its `execution_start_time` frozen. After
-`stuck-sending-recovery-threshold` any pod's next pass picks it up as a
-candidate again and retries. Repeated crashes therefore count against
-`max-attempts` and eventually escalate the row to `ERROR` rather than looping forever.
+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

@@ -22,6 +22,7 @@
 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;
@@ -57,15 +58,17 @@ public ManageEmail manageEmail(
             AttachmentDataSource attachmentDataSource,
             EmailMapper emailMapper,
             @Value("${aboutbits.emailservice.scheduling.max-attempts:3}") int maxAttempts,
-            @Value("${aboutbits.emailservice.scheduling.interval:30000}") long schedulerIntervalMillis
+            @Value("${aboutbits.emailservice.scheduling.interval:30000}") long schedulerIntervalMillis,
+            PlatformTransactionManager transactionManager
     ) {
         return new ManageEmail(
                 emailRepository,
                 javaMailSender,
                 attachmentDataSource,
                 emailMapper,
                 maxAttempts,
-                Duration.ofMillis(schedulerIntervalMillis)
+                Duration.ofMillis(schedulerIntervalMillis),
+                transactionManager
         );
     }
 
@@ -75,7 +78,7 @@ public SendScheduledEmails sendScheduledEmails(
             QueryEmail queryEmail,
             ManageEmail manageEmail,
             List<EmailSchedulerCallback> callbacks,
-            @Value("${aboutbits.emailservice.scheduling.stuck-sending-recovery-threshold:PT5M}") Duration stuckSendingRecoveryThreshold
+            @Value("${aboutbits.emailservice.scheduling.stuck-sending-recovery-threshold:PT30M}") Duration stuckSendingRecoveryThreshold
     ) {
         return new SendScheduledEmails(queryEmail, manageEmail, callbacks, stuckSendingRecoveryThreshold);
     }

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

@@ -18,7 +18,10 @@
 import org.springframework.core.io.ByteArrayResource;
 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;
@@ -40,21 +43,26 @@ public class ManageEmail {
     private final EmailMapper emailMapper;
     private final int maxAttempts;
     private final Duration schedulerInterval;
+    private final TransactionTemplate transactionTemplate;
 
     public ManageEmail(
             EmailRepository emailRepository,
             JavaMailSender mailSender,
             AttachmentDataSource attachmentDataSource,
             EmailMapper emailMapper,
             int maxAttempts,
-            Duration schedulerInterval
+            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 {
@@ -78,10 +86,8 @@ public EmailDto sendOrFail(@Valid EmailParameter parameter) throws EmailExceptio
             throw new EmailException(e);
         }
 
-        email.setState(EmailState.SENDING);
         email.setExecutionStartTime(OffsetDateTime.now());
         email.incrementAttempts();
-        emailRepository.save(email);
 
         try {
             sendMail(email);
@@ -94,7 +100,7 @@ public EmailDto sendOrFail(@Valid EmailParameter parameter) throws EmailExceptio
             email.setErrorMessage(e.getMessage());
         }
 
-        var savedEmail = emailRepository.save(email);
+        var savedEmail = transactionTemplate.execute(_ -> emailRepository.save(email));
 
         if (savedEmail.getState() == EmailState.ERROR) {
             throw new EmailException("Failed to send email [id=%s, providerMessage=%s]"
@@ -121,14 +127,15 @@ Email completeClaimedSend(Email email) {
         } catch (Exception e) {
             email.setExecutionEndTime(OffsetDateTime.now());
             email.setErrorMessage(e.getMessage());
-            if (email.getAttempts() > maxAttempts) {
+            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(email.getAttempts() * 2L))
+                        OffsetDateTime.now()
+                                .plus(schedulerInterval.multipliedBy((long) Math.pow(2, email.getAttempts())))
                 );
             }
         }
@@ -199,14 +206,13 @@ private void sendMail(
             payload.close();
         }
 
-
         mailSender.send(message);
     }
 
     private Email fromParameter(EmailParameter parameter) throws AttachmentException {
         var emailData = parameter.email();
 
-        final var email = new Email();
+        var email = new Email();
         email.setState(EmailState.PENDING);
         email.setScheduledAt(parameter.scheduledAt());
         email.setContent(new EmailContent(

src/main/resources/META-INF/additional-spring-configuration-metadata.json

@@ -27,13 +27,13 @@
         {
             "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.",
-            "defaultValue": "PT5M"
+            "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.",
+            "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
         }
     ]

src/test/java/it/aboutbits/springboot/emailservice/lib/application/SendScheduledEmailsTest.java

@@ -81,7 +81,7 @@ void givenRetryableFailure_sendEmails_shouldRescheduleWithBackoffAndIncrementAtt
                 .satisfies(email -> {
                     assertThat(email.getState()).isEqualTo(EmailState.PENDING);
                     assertThat(email.getAttempts()).isEqualTo(1);
-                    // First retry backoff: attempts (1) * 2 * scheduler interval.
+                    // 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");
@@ -94,7 +94,7 @@ void givenAttemptsAlreadyAtBudget_sendEmails_shouldEscalateToError() {
         doThrow(new MailSendException("smtp down")).when(javaMailSender).send(any(MimeMessage.class));
         emailRepository.save(
                 EmailFactory.once()
-                        .attempts(MAX_ATTEMPTS)
+                        .attempts(MAX_ATTEMPTS - 1)
                         .scheduledAt(OffsetDateTime.now().minusSeconds(30))
                         .build()
         );
@@ -104,10 +104,10 @@ void givenAttemptsAlreadyAtBudget_sendEmails_shouldEscalateToError() {
         assertThat(emailRepository.findAll())
                 .singleElement()
                 .satisfies(email -> {
-                    // The atomic claim UPDATE incremented attempts from MAX_ATTEMPTS to MAX_ATTEMPTS+1,
-                    // which is > threshold, so the failure escalates to ERROR.
+                    // 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 + 1);
+                    assertThat(email.getAttempts()).isEqualTo(MAX_ATTEMPTS);
                     assertThat(email.getErrorMessage()).contains("smtp down");
                 });
     }
@@ -117,7 +117,7 @@ void givenErrorRow_sendEmails_shouldNotRetry() {
         emailRepository.save(
                 EmailFactory.once()
                         .state(EmailState.ERROR)
-                        .attempts(MAX_ATTEMPTS + 1)
+                        .attempts(MAX_ATTEMPTS)
                         .scheduledAt(OffsetDateTime.now().minusMinutes(1))
                         .errorMessage("previous permanent failure")
                         .build()
@@ -129,7 +129,7 @@ void givenErrorRow_sendEmails_shouldNotRetry() {
                 .singleElement()
                 .satisfies(email -> {
                     assertThat(email.getState()).isEqualTo(EmailState.ERROR);
-                    assertThat(email.getAttempts()).isEqualTo(MAX_ATTEMPTS + 1);
+                    assertThat(email.getAttempts()).isEqualTo(MAX_ATTEMPTS);
                 });
         verify(javaMailSender, times(0)).send(any(MimeMessage.class));
     }