make the scheduled attachment cleanup multi pod save

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 transactionTemplate.execute(_ -> 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);
 }

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

@@ -0,0 +1,142 @@
+package it.aboutbits.springboot.emailservice.lib.application;
+
+import it.aboutbits.springboot.emailservice.lib.AttachmentDataSource;
+import it.aboutbits.springboot.emailservice.lib.EmailState;
+import it.aboutbits.springboot.emailservice.lib.exception.AttachmentException;
+import it.aboutbits.springboot.emailservice.lib.jpa.EmailRepository;
+import it.aboutbits.springboot.emailservice.lib.model.Email;
+import it.aboutbits.springboot.emailservice.lib.model.EmailAttachment;
+import it.aboutbits.springboot.emailservice.support.database.WithPostgres;
+import it.aboutbits.springboot.emailservice.support.database.factory.EmailFactory;
+import org.jspecify.annotations.NullMarked;
+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.util.HashSet;
+import java.util.Set;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.Executors;
+import java.util.concurrent.TimeUnit;
+import java.util.stream.IntStream;
+
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.mockito.ArgumentMatchers.anyLong;
+import static org.mockito.Mockito.doThrow;
+import static org.mockito.Mockito.times;
+import static org.mockito.Mockito.verify;
+
+@SpringBootTest(properties = {
+        "aboutbits.emailservice.scheduling.interval=30000"
+})
+@WithPostgres
+@NullMarked
+class CleanupAttachmentFilesTest {
+    @MockitoBean
+    AttachmentDataSource attachmentDataSource;
+
+    @Autowired
+    EmailRepository emailRepository;
+
+    @Autowired
+    CleanupAttachmentFiles cleanupAttachmentFiles;
+
+    @Test
+    void givenSentUncleanedEmails_cleanupAttachments_shouldReleaseAndMarkCleaned() throws AttachmentException {
+        persistCleanableEmail(100L);
+        persistCleanableEmail(101L);
+        persistCleanableEmail(102L);
+
+        cleanupAttachmentFiles.cleanupAttachments();
+
+        assertThat(emailRepository.findAll())
+                .hasSize(3)
+                .allMatch(Email::isAttachmentsCleaned);
+        verify(attachmentDataSource, times(1)).releaseAttachment(100L);
+        verify(attachmentDataSource, times(1)).releaseAttachment(101L);
+        verify(attachmentDataSource, times(1)).releaseAttachment(102L);
+    }
+
+    @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);
+
+        cleanupAttachmentFiles.cleanupAttachments();
+
+        verify(attachmentDataSource, times(0)).releaseAttachment(anyLong());
+        assertThat(emailRepository.findAll())
+                .filteredOn(email -> email.getState() != EmailState.SENT)
+                .allMatch(email -> !email.isAttachmentsCleaned());
+    }
+
+    @Test
+    void givenReleaseFails_cleanupAttachments_shouldResetFlagForRetry() throws AttachmentException {
+        var email = persistCleanableEmail(300L);
+        doThrow(new AttachmentException()).when(attachmentDataSource).releaseAttachment(300L);
+
+        cleanupAttachmentFiles.cleanupAttachments();
+
+        assertThat(emailRepository.findById(email.getId()))
+                .get()
+                .satisfies(reloaded -> assertThat(reloaded.isAttachmentsCleaned()).isFalse());
+    }
+
+    @Test
+    void givenConcurrentPasses_cleanupAttachments_shouldReleaseEachEmailExactlyOnce() throws Exception {
+        var totalEmails = 25;
+        var workerThreads = 4;
+        var passesPerWorker = 10;
+        for (var i = 0; i < totalEmails; i++) {
+            persistCleanableEmail(1000L + i);
+        }
+
+        var startGate = new CountDownLatch(1);
+        var executor = Executors.newFixedThreadPool(workerThreads);
+        try {
+            IntStream.range(0, workerThreads).forEach(_ -> executor.submit(() -> {
+                startGate.await();
+                for (var i = 0; i < passesPerWorker; i++) {
+                    cleanupAttachmentFiles.cleanupAttachments();
+                }
+                return null;
+            }));
+            startGate.countDown();
+            executor.shutdown();
+            assertThat(executor.awaitTermination(60, TimeUnit.SECONDS)).isTrue();
+        } finally {
+            if (!executor.isTerminated()) {
+                executor.shutdownNow();
+            }
+        }
+
+        assertThat(emailRepository.findAll())
+                .hasSize(totalEmails)
+                .allMatch(Email::isAttachmentsCleaned);
+        // Exactly one release per file reference
+        verify(attachmentDataSource, times(totalEmails)).releaseAttachment(anyLong());
+    }
+
+    private Email persistCleanableEmail(long fileReference) {
+        return persistEmail(fileReference, EmailState.SENT, false);
+    }
+
+    private Email persistEmail(long fileReference, EmailState state, boolean attachmentsCleaned) {
+        var email = EmailFactory.once()
+                .state(state)
+                .attachmentsCleaned(attachmentsCleaned)
+                .build();
+
+        var attachment = new EmailAttachment();
+        attachment.setEmail(email);
+        attachment.setFileName("file.png");
+        attachment.setContentType("image/png");
+        attachment.setFileReference(fileReference);
+        email.setAttachments(new HashSet<>(Set.of(attachment)));
+
+        return emailRepository.save(email);
+    }
+}