make the scheduled attachment cleanup multip pod save

Author: J0nasMayr

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

@@ -29,15 +29,24 @@ public class CleanupAttachmentFiles {
     void cleanupAttachments() {
         logStartOfPass();
 
-        var emailsToCleanup = queryEmail.readyToCleanup();
+        var candidateIds = queryEmail.candidateIdsToCleanup();
 
+        var countClaimed = 0;
         var countCleaned = 0;
         var countError = 0;
-        for (var email : emailsToCleanup) {
+        for (var id : candidateIds) {
+            var claimed = manageEmail.tryClaimForCleanup(id);
+
+            if (claimed.isEmpty()) {
+                // Lost race to another pod; Skip
+                continue;
+            }
+            countClaimed++;
+
             try {
-                manageEmail.cleanupAttachments(email);
+                manageEmail.completeClaimedCleanup(claimed.get());
                 countCleaned++;
-            } catch (AttachmentException e) {
+            } catch (AttachmentException _) {
                 countError++;
             }
         }
@@ -46,7 +55,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/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,25 @@ Email completeClaimedSend(Email email) {
         return emailRepository.save(email);
     }
 
-    void cleanupAttachments(final Email email) throws AttachmentException {
-        for (var attachment : email.getAttachments()) {
-            attachmentDataSource.releaseAttachment(attachment.getFileReference());
+    // Atomically flips a not-yet-cleaned SENT row to cleaned
+    @Transactional
+    Optional<Email> tryClaimForCleanup(long id) {
+        var claimed = emailRepository.claimForCleanup(id);
+        return claimed == 0 ? Optional.empty() : emailRepository.findById(id);
+    }
+
+    // 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;
         }
-        email.setAttachmentsCleaned(true);
-        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() {
+        return emailRepository.findCandidateIdsToCleanup();
     }
 
     public Optional<EmailDto> byId(long id) {

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

@@ -84,11 +84,24 @@ 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.
     @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
             """)
-    List<Email> findReadyToCleanup();
+    List<Long> findCandidateIdsToCleanup();
+
+    // Atomic compare-and-set claim: flips a single not-yet-cleaned SENT row to cleaned.
+    // 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
+               where e.id = :id
+                   and e.attachmentsCleaned = false
+                   and e.state = it.aboutbits.springboot.emailservice.lib.EmailState.SENT
+            """)
+    int claimForCleanup(@Param("id") long id);
 }