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,13 +30,14 @@ public class CleanupAttachmentFiles {
     void cleanupAttachments() {
         logStartOfPass();
 
-        var candidateIds = queryEmail.candidateIdsToCleanup();
+        var staleCleanupBefore = OffsetDateTime.now().minus(stuckCleanupRecoveryThreshold);
+        var candidateIds = queryEmail.candidateIdsToCleanup(staleCleanupBefore);
 
         var countClaimed = 0;
         var countCleaned = 0;
         var countError = 0;
         for (var id : candidateIds) {
-            var claimed = manageEmail.tryClaimForCleanup(id);
+            var claimed = manageEmail.tryClaimForCleanup(id, staleCleanupBefore);
 
             if (claimed.isEmpty()) {
                 // Lost race to another pod; Skip
@@ -46,7 +48,7 @@ void cleanupAttachments() {
             try {
                 manageEmail.completeClaimedCleanup(claimed.get());
                 countCleaned++;
-            } catch (AttachmentException _) {
+            } catch (Exception _) {
                 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

@@ -142,25 +142,20 @@ Email completeClaimedSend(Email email) {
         return transactionTemplate.execute(_ -> emailRepository.save(email));
     }
 
-    // Atomically flips a not-yet-cleaned SENT row to cleaned
+    // Atomically marks a not-yet-cleaned SENT row as cleanup-in-progress
     @Transactional
-    Optional<Email> tryClaimForCleanup(long id) {
-        var claimed = emailRepository.claimForCleanup(id);
+    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
+    // Actually release the attachment payloads of the claimed email.
     void completeClaimedCleanup(final Email email) throws AttachmentException {
-        try {
-            for (var attachment : email.getAttachments()) {
-                attachmentDataSource.releaseAttachment(attachment.getFileReference());
-            }
-        } catch (AttachmentException e) {
-            // Undo the claim so the row is retried on a later pass
-            email.setAttachmentsCleaned(false);
-            transactionTemplate.execute(_ -> emailRepository.save(email));
-            throw e;
+        for (var attachment : email.getAttachments()) {
+            attachmentDataSource.releaseAttachment(attachment.getFileReference());
         }
+        email.setAttachmentsCleaned(true);
+        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

@@ -46,8 +46,8 @@ List<Long> candidateIdsToSend(OffsetDateTime staleSendingBefore) {
         );
     }
 
-    List<Long> candidateIdsToCleanup() {
-        return emailRepository.findCandidateIdsToCleanup();
+    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

@@ -86,22 +86,36 @@ int claimForSend(
 
     // 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.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<Long> findCandidateIdsToCleanup();
+    List<Long> findCandidateIdsToCleanup(@Param("staleCleanupBefore") OffsetDateTime staleCleanupBefore);
 
-    // Atomic compare-and-set claim: flips a single not-yet-cleaned SENT row to cleaned.
+    // 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.attachmentsCleaned = true
+               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);
+    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",

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

@@ -1,5 +1,6 @@
 package it.aboutbits.springboot.emailservice.lib.application;
 
+import it.aboutbits.springboot.emailservice.lib.AttachmentCleanerCallback;
 import it.aboutbits.springboot.emailservice.lib.AttachmentDataSource;
 import it.aboutbits.springboot.emailservice.lib.EmailState;
 import it.aboutbits.springboot.emailservice.lib.exception.AttachmentException;
@@ -9,11 +10,14 @@
 import it.aboutbits.springboot.emailservice.support.database.WithPostgres;
 import it.aboutbits.springboot.emailservice.support.database.factory.EmailFactory;
 import org.jspecify.annotations.NullMarked;
+import org.jspecify.annotations.Nullable;
 import org.junit.jupiter.api.Test;
 import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.boot.test.context.SpringBootTest;
 import org.springframework.test.context.bean.override.mockito.MockitoBean;
 
+import java.time.OffsetDateTime;
+import java.time.temporal.ChronoUnit;
 import java.util.HashSet;
 import java.util.Set;
 import java.util.concurrent.CountDownLatch;
@@ -28,6 +32,7 @@
 import static org.mockito.Mockito.verify;
 
 @SpringBootTest(properties = {
+        "aboutbits.emailservice.scheduling.stuck-cleanup-recovery-threshold=PT5M",
         "aboutbits.emailservice.scheduling.interval=30000"
 })
 @WithPostgres
@@ -36,6 +41,9 @@ class CleanupAttachmentFilesTest {
     @MockitoBean
     AttachmentDataSource attachmentDataSource;
 
+    @MockitoBean
+    AttachmentCleanerCallback attachmentCleanerCallback;
+
     @Autowired
     EmailRepository emailRepository;
 
@@ -56,14 +64,15 @@ void givenSentUncleanedEmails_cleanupAttachments_shouldReleaseAndMarkCleaned() t
         verify(attachmentDataSource, times(1)).releaseAttachment(100L);
         verify(attachmentDataSource, times(1)).releaseAttachment(101L);
         verify(attachmentDataSource, times(1)).releaseAttachment(102L);
+        verify(attachmentCleanerCallback).report(new AttachmentCleanerCallback.Report(3, 3, 0));
     }
 
     @Test
     void givenNonSentOrAlreadyCleanedRows_cleanupAttachments_shouldSkip() throws AttachmentException {
-        persistEmail(200L, EmailState.PENDING, false);
-        persistEmail(201L, EmailState.SENDING, false);
-        persistEmail(202L, EmailState.ERROR, false);
-        persistEmail(203L, EmailState.SENT, true);
+        persistEmail(200L, EmailState.PENDING, false, null);
+        persistEmail(201L, EmailState.SENDING, false, null);
+        persistEmail(202L, EmailState.ERROR, false, null);
+        persistEmail(203L, EmailState.SENT, true, null);
 
         cleanupAttachmentFiles.cleanupAttachments();
 
@@ -74,15 +83,50 @@ void givenNonSentOrAlreadyCleanedRows_cleanupAttachments_shouldSkip() throws Att
     }
 
     @Test
-    void givenReleaseFails_cleanupAttachments_shouldResetFlagForRetry() throws AttachmentException {
+    void givenReleaseFails_cleanupAttachments_shouldLeaveClaimForStaleRecovery() throws AttachmentException {
         var email = persistCleanableEmail(300L);
         doThrow(new AttachmentException()).when(attachmentDataSource).releaseAttachment(300L);
 
         cleanupAttachmentFiles.cleanupAttachments();
 
+        // Files were not released, and the claim is left in place so a later pass recovers it
+        assertThat(emailRepository.findById(email.getId()))
+                .get()
+                .satisfies(reloaded -> {
+                    assertThat(reloaded.isAttachmentsCleaned()).isFalse();
+                    assertThat(reloaded.getCleanupStartTime()).isNotNull();
+                });
+        verify(attachmentCleanerCallback).report(new AttachmentCleanerCallback.Report(1, 0, 1));
+    }
+
+    @Test
+    void givenStuckCleanup_cleanupAttachments_shouldRecoverAndClean() throws AttachmentException {
+        var staleStart = OffsetDateTime.now().minusMinutes(10);
+        var email = persistEmail(400L, EmailState.SENT, false, staleStart);
+
+        cleanupAttachmentFiles.cleanupAttachments();
+
         assertThat(emailRepository.findById(email.getId()))
                 .get()
-                .satisfies(reloaded -> assertThat(reloaded.isAttachmentsCleaned()).isFalse());
+                .satisfies(reloaded -> assertThat(reloaded.isAttachmentsCleaned()).isTrue());
+        verify(attachmentDataSource, times(1)).releaseAttachment(400L);
+    }
+
+    @Test
+    void givenFreshCleanupInProgress_cleanupAttachments_shouldNotStealFromOtherPod() throws AttachmentException {
+        // cleanupStartTime within the 5-minute threshold -> another pod is cleaning this right now -> do not re-claim
+        var recentStart = OffsetDateTime.now().minusSeconds(30).truncatedTo(ChronoUnit.MICROS);
+        var email = persistEmail(500L, EmailState.SENT, false, recentStart);
+
+        cleanupAttachmentFiles.cleanupAttachments();
+
+        assertThat(emailRepository.findById(email.getId()))
+                .get()
+                .satisfies(reloaded -> {
+                    assertThat(reloaded.isAttachmentsCleaned()).isFalse();
+                    assertThat(reloaded.getCleanupStartTime()).isEqualTo(recentStart);
+                });
+        verify(attachmentDataSource, times(0)).releaseAttachment(anyLong());
     }
 
     @Test
@@ -121,13 +165,19 @@ void givenConcurrentPasses_cleanupAttachments_shouldReleaseEachEmailExactlyOnce(
     }
 
     private Email persistCleanableEmail(long fileReference) {
-        return persistEmail(fileReference, EmailState.SENT, false);
+        return persistEmail(fileReference, EmailState.SENT, false, null);
     }
 
-    private Email persistEmail(long fileReference, EmailState state, boolean attachmentsCleaned) {
+    private Email persistEmail(
+            long fileReference,
+            EmailState state,
+            boolean attachmentsCleaned,
+            @Nullable OffsetDateTime cleanupStartTime
+    ) {
         var email = EmailFactory.once()
                 .state(state)
                 .attachmentsCleaned(attachmentsCleaned)
+                .cleanupStartTime(cleanupStartTime)
                 .build();
 
         var attachment = new EmailAttachment();