make the scheduled attachment cleanup multi pod save (#13)

* first draft for concurrency using an approach at the database level

* Revert "first draft for concurrency using an approach at the database level"

This reverts commit 15c22d4.

* refactor Email entity by embedding EmailContent and adding new fields

* solve concurrency issue between pods using atomic compare-and-set; Add new configurables and make sendOrFail not retry

* add test for concurrency and schedule functionality

* update readme.md

* test fixes

* truncate OffsetDateTime to MICROS for pipeline to run successfully

* remove unnecessary errorMessage set

* requested changes

* consider all error cases that can happen on a schedule pass

* remove the first try catch since it is overkill

* make the scheduled attachment cleanup multip pod save

* add tests

* requested changes

Author: J0nasMayr

readme.md

@@ -81,12 +81,14 @@ The following configuration options are available:
 | `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.stuck-cleanup-recovery-threshold`    | PT30M   | How long an email may keep its attachment-cleanup lock before the cleanup is considered abandoned (crashed pod) and eligible to be re-claimed. Must comfortably exceed the worst-case duration of releasing all attachments of a single email, otherwise a slow in-flight cleanup can be re-claimed by another pod and its attachments released twice (harmless only if `AttachmentDataSource.releaseAttachment` is idempotent). |
 | `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.
+Both schedulers are safe to run on every pod concurrently: the database arbitrates which pod handles 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.
+Attachment Cleanup is at-least-once - if a pod crashes after releasing the attachments but before the result was persisted, the cleanup may be retried on recovery.
 
 ## Local development:
 

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

@@ -88,9 +88,10 @@ public SendScheduledEmails sendScheduledEmails(
     public CleanupAttachmentFiles cleanupAttachments(
             QueryEmail queryEmail,
             ManageEmail manageEmail,
-            List<AttachmentCleanerCallback> callbacks
+            List<AttachmentCleanerCallback> callbacks,
+            @Value("${aboutbits.emailservice.scheduling.stuck-cleanup-recovery-threshold:PT30M}") Duration stuckCleanupRecoveryThreshold
     ) {
-        return new CleanupAttachmentFiles(queryEmail, manageEmail, callbacks);
+        return new CleanupAttachmentFiles(queryEmail, manageEmail, callbacks, stuckCleanupRecoveryThreshold);
     }
 
     @Bean

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

@@ -2,13 +2,13 @@
 
 
 import it.aboutbits.springboot.emailservice.lib.AttachmentCleanerCallback;
-import it.aboutbits.springboot.emailservice.lib.exception.AttachmentException;
 import lombok.RequiredArgsConstructor;
 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;
 
 @RequiredArgsConstructor
@@ -20,6 +20,7 @@ public class CleanupAttachmentFiles {
     private final QueryEmail queryEmail;
     private final ManageEmail manageEmail;
     private final List<AttachmentCleanerCallback> callbacks;
+    private final Duration stuckCleanupRecoveryThreshold;
 
     private long lastInfoLogMillis = System.currentTimeMillis();
     private long silentRuns = 0;
@@ -29,15 +30,25 @@ public class CleanupAttachmentFiles {
     void cleanupAttachments() {
         logStartOfPass();
 
-        var emailsToCleanup = queryEmail.readyToCleanup();
+        var staleCleanupBefore = OffsetDateTime.now().minus(stuckCleanupRecoveryThreshold);
+        var candidateIds = queryEmail.candidateIdsToCleanup(staleCleanupBefore);
 
+        var countClaimed = 0;
         var countCleaned = 0;
         var countError = 0;
-        for (var email : emailsToCleanup) {
+        for (var id : candidateIds) {
+            var claimed = manageEmail.tryClaimForCleanup(id, staleCleanupBefore);
+
+            if (claimed.isEmpty()) {
+                // Lost race to another pod; Skip
+                continue;
+            }
+            countClaimed++;
+
             try {
-                manageEmail.cleanupAttachments(email);
+                manageEmail.completeClaimedCleanup(claimed.get());
                 countCleaned++;
-            } catch (AttachmentException e) {
+            } catch (Exception _) {
                 countError++;
             }
         }
@@ -46,7 +57,7 @@ void cleanupAttachments() {
 
         for (var callback : callbacks) {
             callback.report(new AttachmentCleanerCallback.Report(
-                    emailsToCleanup.size(),
+                    countClaimed,
                     countCleaned,
                     countError
             ));

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

@@ -74,6 +74,8 @@ updated_at           timestamp with time zone default now()           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;
 
+                 alter table email_service_emails add column if not exists cleanup_start_time timestamp with time zone;
+
                  update email_service_emails
                     set execution_end_time = sent_at
                   where sent_at is not null

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

@@ -60,7 +60,7 @@ public ManageEmail(
         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
+        // Persist the final email state in its own, independent transaction to never make it roll back
         this.transactionTemplate = new TransactionTemplate(transactionManager);
         this.transactionTemplate.setPropagationBehavior(TransactionDefinition.PROPAGATION_REQUIRES_NEW);
     }
@@ -142,12 +142,20 @@ Email completeClaimedSend(Email email) {
         return transactionTemplate.execute(_ -> emailRepository.save(email));
     }
 
-    void cleanupAttachments(final Email email) throws AttachmentException {
+    // Atomically marks a not-yet-cleaned SENT row as cleanup-in-progress
+    @Transactional
+    Optional<Email> tryClaimForCleanup(long id, OffsetDateTime staleCleanupBefore) {
+        var claimed = emailRepository.claimForCleanup(id, OffsetDateTime.now(), staleCleanupBefore);
+        return claimed == 0 ? Optional.empty() : emailRepository.findById(id);
+    }
+
+    // Actually release the attachment payloads of the claimed email.
+    void completeClaimedCleanup(final Email email) throws AttachmentException {
         for (var attachment : email.getAttachments()) {
             attachmentDataSource.releaseAttachment(attachment.getFileReference());
         }
         email.setAttachmentsCleaned(true);
-        emailRepository.save(email);
+        transactionTemplate.execute(_ -> emailRepository.save(email));
     }
 
     private void sendMail(Email email) throws MessagingException, IOException, AttachmentException {

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

@@ -4,7 +4,6 @@
 import it.aboutbits.springboot.emailservice.lib.EmailDto;
 import it.aboutbits.springboot.emailservice.lib.EmailState;
 import it.aboutbits.springboot.emailservice.lib.jpa.EmailRepository;
-import it.aboutbits.springboot.emailservice.lib.model.Email;
 import lombok.RequiredArgsConstructor;
 import org.jspecify.annotations.NullMarked;
 import org.springframework.data.domain.Page;
@@ -47,8 +46,8 @@ List<Long> candidateIdsToSend(OffsetDateTime staleSendingBefore) {
         );
     }
 
-    List<Email> readyToCleanup() {
-        return emailRepository.findReadyToCleanup();
+    List<Long> candidateIdsToCleanup(OffsetDateTime staleCleanupBefore) {
+        return emailRepository.findCandidateIdsToCleanup(staleCleanupBefore);
     }
 
     public Optional<EmailDto> byId(long id) {

src/main/java/it/aboutbits/springboot/emailservice/lib/jpa/EmailRepository.java

@@ -84,11 +84,38 @@ int claimForSend(
             @Param("staleSendingBefore") OffsetDateTime staleSendingBefore
     );
 
-    @EntityGraph(value = Email.DEFAULT_ENTITY_GRAPH)
+    // Plain read, no locking -> two pods may see overlapping candidate sets.
+    // The atomic UPDATE in claimForCleanup arbitrates the actual claim.
+    // Includes rows whose cleanup was abandoned by a crashed pod (past the stale threshold).
     @Query("""
-            select e from Email e
+            select e.id from Email e
                 where e.attachmentsCleaned = false
                     and e.state = it.aboutbits.springboot.emailservice.lib.EmailState.SENT
+                    and (
+                        e.cleanupStartTime is null
+                        or e.cleanupStartTime < :staleCleanupBefore
+                    )
             """)
-    List<Email> findReadyToCleanup();
+    List<Long> findCandidateIdsToCleanup(@Param("staleCleanupBefore") OffsetDateTime staleCleanupBefore);
+
+    // Atomic compare-and-set claim: marks a single not-yet-cleaned SENT row as cleanup-in-progress
+    // by stamping cleanupStartTime, if it is claimable (never started, or 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.cleanupStartTime = :now
+               where e.id = :id
+                   and e.attachmentsCleaned = false
+                   and e.state = it.aboutbits.springboot.emailservice.lib.EmailState.SENT
+                   and (
+                       e.cleanupStartTime is null
+                       or e.cleanupStartTime < :staleCleanupBefore
+                   )
+            """)
+    int claimForCleanup(
+            @Param("id") long id,
+            @Param("now") OffsetDateTime now,
+            @Param("staleCleanupBefore") OffsetDateTime staleCleanupBefore
+    );
 }

src/main/java/it/aboutbits/springboot/emailservice/lib/model/Email.java

@@ -58,6 +58,9 @@ public class Email {
     @Builder.Default
     private boolean attachmentsCleaned = false;
 
+    @Nullable
+    private OffsetDateTime cleanupStartTime;
+
     private OffsetDateTime scheduledAt;
     @Nullable
     private OffsetDateTime executionStartTime;

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

@@ -30,6 +30,12 @@
             "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.stuck-cleanup-recovery-threshold",
+            "type": "java.time.Duration",
+            "description": "How long an email may keep its attachment-cleanup lock before the cleanup is considered abandoned (crashed pod) and eligible to be re-claimed. Must comfortably exceed the worst-case duration of releasing all attachments of a single email, otherwise a slow in-flight cleanup can be re-claimed by another pod and its attachments released twice (harmless only if `AttachmentDataSource.releaseAttachment` is idempotent).",
+            "defaultValue": "PT30M"
+        },
         {
             "name": "aboutbits.emailservice.scheduling.max-attempts",
             "type": "java.lang.Integer",