first draft for concurrency using an approach at the database level

readme.md

@@ -74,11 +74,40 @@ 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`    | 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.        |
+
+## 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.
 
 ## Local development:
 

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

@@ -14,15 +14,16 @@
 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 java.time.Duration;
 import java.util.List;
 
 @AutoConfigurationPackage
@@ -45,24 +46,47 @@ 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
+    ) {
+        return new ManageEmail(
+                emailRepository,
+                javaMailSender,
+                attachmentDataSource,
+                emailMapper,
+                maxAttempts,
+                Duration.ofMillis(schedulerIntervalMillis)
+        );
     }
 
     @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:PT5M}") 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
         );