diff --git a/readme.md b/readme.md index cf86fff..f634ff9 100644 --- a/readme.md +++ b/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: diff --git a/src/main/java/it/aboutbits/springboot/emailservice/EmailServiceConfiguration.java b/src/main/java/it/aboutbits/springboot/emailservice/EmailServiceConfiguration.java index c882776..0c4bc36 100644 --- a/src/main/java/it/aboutbits/springboot/emailservice/EmailServiceConfiguration.java +++ b/src/main/java/it/aboutbits/springboot/emailservice/EmailServiceConfiguration.java @@ -88,9 +88,10 @@ public SendScheduledEmails sendScheduledEmails( public CleanupAttachmentFiles cleanupAttachments( QueryEmail queryEmail, ManageEmail manageEmail, - List callbacks + List 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 diff --git a/src/main/java/it/aboutbits/springboot/emailservice/lib/application/CleanupAttachmentFiles.java b/src/main/java/it/aboutbits/springboot/emailservice/lib/application/CleanupAttachmentFiles.java index df7209d..95cbf16 100644 --- a/src/main/java/it/aboutbits/springboot/emailservice/lib/application/CleanupAttachmentFiles.java +++ b/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 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 )); diff --git a/src/main/java/it/aboutbits/springboot/emailservice/lib/application/EmailServiceMigrator.java b/src/main/java/it/aboutbits/springboot/emailservice/lib/application/EmailServiceMigrator.java index 919411a..722e745 100644 --- a/src/main/java/it/aboutbits/springboot/emailservice/lib/application/EmailServiceMigrator.java +++ b/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 diff --git a/src/main/java/it/aboutbits/springboot/emailservice/lib/application/ManageEmail.java b/src/main/java/it/aboutbits/springboot/emailservice/lib/application/ManageEmail.java index 7563888..12cc6ce 100644 --- a/src/main/java/it/aboutbits/springboot/emailservice/lib/application/ManageEmail.java +++ b/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 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 { diff --git a/src/main/java/it/aboutbits/springboot/emailservice/lib/application/QueryEmail.java b/src/main/java/it/aboutbits/springboot/emailservice/lib/application/QueryEmail.java index 36da5bf..dad15c7 100644 --- a/src/main/java/it/aboutbits/springboot/emailservice/lib/application/QueryEmail.java +++ b/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 candidateIdsToSend(OffsetDateTime staleSendingBefore) { ); } - List readyToCleanup() { - return emailRepository.findReadyToCleanup(); + List candidateIdsToCleanup(OffsetDateTime staleCleanupBefore) { + return emailRepository.findCandidateIdsToCleanup(staleCleanupBefore); } public Optional byId(long id) { diff --git a/src/main/java/it/aboutbits/springboot/emailservice/lib/jpa/EmailRepository.java b/src/main/java/it/aboutbits/springboot/emailservice/lib/jpa/EmailRepository.java index 4c8c150..f77d0d1 100644 --- a/src/main/java/it/aboutbits/springboot/emailservice/lib/jpa/EmailRepository.java +++ b/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 findReadyToCleanup(); + List 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 + ); } diff --git a/src/main/java/it/aboutbits/springboot/emailservice/lib/model/Email.java b/src/main/java/it/aboutbits/springboot/emailservice/lib/model/Email.java index 2c09720..630852d 100644 --- a/src/main/java/it/aboutbits/springboot/emailservice/lib/model/Email.java +++ b/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; diff --git a/src/main/resources/META-INF/additional-spring-configuration-metadata.json b/src/main/resources/META-INF/additional-spring-configuration-metadata.json index 7b42404..b180d13 100644 --- a/src/main/resources/META-INF/additional-spring-configuration-metadata.json +++ b/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", diff --git a/src/test/java/it/aboutbits/springboot/emailservice/lib/application/CleanupAttachmentFilesTest.java b/src/test/java/it/aboutbits/springboot/emailservice/lib/application/CleanupAttachmentFilesTest.java new file mode 100644 index 0000000..c6eeb66 --- /dev/null +++ b/src/test/java/it/aboutbits/springboot/emailservice/lib/application/CleanupAttachmentFilesTest.java @@ -0,0 +1,192 @@ +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; +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.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; +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.stuck-cleanup-recovery-threshold=PT5M", + "aboutbits.emailservice.scheduling.interval=30000" +}) +@WithPostgres +@NullMarked +class CleanupAttachmentFilesTest { + @MockitoBean + AttachmentDataSource attachmentDataSource; + + @MockitoBean + AttachmentCleanerCallback attachmentCleanerCallback; + + @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); + verify(attachmentCleanerCallback).report(new AttachmentCleanerCallback.Report(3, 3, 0)); + } + + @Test + void givenNonSentOrAlreadyCleanedRows_cleanupAttachments_shouldSkip() throws AttachmentException { + 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(); + + verify(attachmentDataSource, times(0)).releaseAttachment(anyLong()); + assertThat(emailRepository.findAll()) + .filteredOn(email -> email.getState() != EmailState.SENT) + .allMatch(email -> !email.isAttachmentsCleaned()); + } + + @Test + 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()).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 + 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, null); + } + + 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(); + 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); + } +}