Skip to content

Commit e37c8a7

Browse files
J0nasMayrJ0nasMayr
authored
first draft for concurrency using an approach at the database level (#12)
* 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 last save be in its on transaction so it can never be rolled back
1 parent 8ef5819 commit e37c8a7

15 files changed

Lines changed: 580 additions & 150 deletions

File tree

‎readme.md‎

Lines changed: 13 additions & 5 deletions
Original file line numberDiff line numberDiff line change
@@ -74,11 +74,19 @@ public class App {
7474

7575
The following configuration options are available:
7676

77-
| Name | Default | Description |
78-
|----------------------------------------|-------------|-----------------------------------------------------------------------|
79-
| `lib.emailservice.migrations.enabled` | true | Enables database migrations. |
80-
| `lib.emailservice.scheduling.enabled` | true | Enables the scheduler sending the emails. |
81-
| `lib.emailservice.scheduling.interval` | 30000 | Specifies the milliseconds delay between runs of the scheduler. |
77+
| Name | Default | Description |
78+
|-------------------------------------------------------------------------|---------|------------------------------------------------------------------------------------------------------------------------|
79+
| `aboutbits.emailservice.migrations.enabled` | true | Enables database migrations. |
80+
| `aboutbits.emailservice.scheduling.enabled` | true | Enables the scheduler sending the emails. |
81+
| `aboutbits.emailservice.scheduling.cleanup.enabled` | true | Enables cleanup of attachment files after sending. |
82+
| `aboutbits.emailservice.scheduling.interval` | 30000 | Milliseconds delay between runs of the scheduler. |
83+
| `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. |
84+
| `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`. |
85+
86+
## Multi-pod deployments
87+
88+
The scheduler is safe to run on every pod concurrently: the database arbitrates which pod sends each email.
89+
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.
8290

8391
## Local development:
8492

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

Lines changed: 35 additions & 8 deletions
Original file line numberDiff line numberDiff line change
@@ -14,15 +14,17 @@
1414
import it.aboutbits.springboot.emailservice.lib.application.SendScheduledEmails;
1515
import it.aboutbits.springboot.emailservice.lib.application.UnavailableAttachmentDataSource;
1616
import it.aboutbits.springboot.emailservice.lib.jpa.EmailRepository;
17-
import jakarta.persistence.EntityManager;
1817
import org.jspecify.annotations.NullMarked;
18+
import org.springframework.beans.factory.annotation.Value;
1919
import org.springframework.boot.autoconfigure.AutoConfigurationPackage;
2020
import org.springframework.boot.autoconfigure.condition.ConditionalOnMissingBean;
2121
import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
2222
import org.springframework.context.annotation.Bean;
2323
import org.springframework.jdbc.core.JdbcTemplate;
2424
import org.springframework.mail.javamail.JavaMailSender;
25+
import org.springframework.transaction.PlatformTransactionManager;
2526

27+
import java.time.Duration;
2628
import java.util.List;
2729

2830
@AutoConfigurationPackage
@@ -45,24 +47,49 @@ public EmailMapper emailMapper() {
4547
}
4648

4749
@Bean
48-
public QueryEmail queryEmail(EmailRepository emailRepository, EmailMapper emailMapper, EntityManager entityManager) {
49-
return new QueryEmail(emailRepository, emailMapper, entityManager);
50+
public QueryEmail queryEmail(EmailRepository emailRepository, EmailMapper emailMapper) {
51+
return new QueryEmail(emailRepository, emailMapper);
5052
}
5153

5254
@Bean
53-
public ManageEmail manageEmail(EmailRepository emailRepository, JavaMailSender javaMailSender, AttachmentDataSource attachmentDataSource, EmailMapper emailMapper) {
54-
return new ManageEmail(emailRepository, javaMailSender, attachmentDataSource, emailMapper);
55+
public ManageEmail manageEmail(
56+
EmailRepository emailRepository,
57+
JavaMailSender javaMailSender,
58+
AttachmentDataSource attachmentDataSource,
59+
EmailMapper emailMapper,
60+
@Value("${aboutbits.emailservice.scheduling.max-attempts:3}") int maxAttempts,
61+
@Value("${aboutbits.emailservice.scheduling.interval:30000}") long schedulerIntervalMillis,
62+
PlatformTransactionManager transactionManager
63+
) {
64+
return new ManageEmail(
65+
emailRepository,
66+
javaMailSender,
67+
attachmentDataSource,
68+
emailMapper,
69+
maxAttempts,
70+
Duration.ofMillis(schedulerIntervalMillis),
71+
transactionManager
72+
);
5573
}
5674

5775
@Bean
5876
@ConditionalOnProperty(value = "aboutbits.emailservice.scheduling.enabled", matchIfMissing = true)
59-
public SendScheduledEmails sendScheduledEmails(QueryEmail queryEmail, ManageEmail manageEmail, List<EmailSchedulerCallback> callbacks) {
60-
return new SendScheduledEmails(queryEmail, manageEmail, callbacks);
77+
public SendScheduledEmails sendScheduledEmails(
78+
QueryEmail queryEmail,
79+
ManageEmail manageEmail,
80+
List<EmailSchedulerCallback> callbacks,
81+
@Value("${aboutbits.emailservice.scheduling.stuck-sending-recovery-threshold:PT30M}") Duration stuckSendingRecoveryThreshold
82+
) {
83+
return new SendScheduledEmails(queryEmail, manageEmail, callbacks, stuckSendingRecoveryThreshold);
6184
}
6285

6386
@Bean
6487
@ConditionalOnProperty(value = "aboutbits.emailservice.scheduling.cleanup.enabled", matchIfMissing = true)
65-
public CleanupAttachmentFiles cleanupAttachments(QueryEmail queryEmail, ManageEmail manageEmail, List<AttachmentCleanerCallback> callbacks) {
88+
public CleanupAttachmentFiles cleanupAttachments(
89+
QueryEmail queryEmail,
90+
ManageEmail manageEmail,
91+
List<AttachmentCleanerCallback> callbacks
92+
) {
6693
return new CleanupAttachmentFiles(queryEmail, manageEmail, callbacks);
6794
}
6895

‎src/main/java/it/aboutbits/springboot/emailservice/lib/EmailDto.java‎

Lines changed: 5 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -32,10 +32,12 @@ public record EmailDto(
3232

3333
OffsetDateTime scheduledAt,
3434
@Nullable
35-
OffsetDateTime sentAt,
36-
35+
OffsetDateTime executionStartTime,
3736
@Nullable
38-
OffsetDateTime errorAt,
37+
OffsetDateTime executionEndTime,
38+
39+
int attempts,
40+
3941
@Nullable
4042
String errorMessage,
4143

‎src/main/java/it/aboutbits/springboot/emailservice/lib/EmailState.java‎

Lines changed: 1 addition & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -5,6 +5,7 @@
55
@NullMarked
66
public enum EmailState {
77
PENDING,
8+
SENDING,
89
SENT,
910
ERROR
1011
}

‎src/main/java/it/aboutbits/springboot/emailservice/lib/application/EmailMapper.java‎

Lines changed: 9 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -6,6 +6,7 @@
66
import org.jspecify.annotations.NullUnmarked;
77
import org.mapstruct.AnnotateWith;
88
import org.mapstruct.Mapper;
9+
import org.mapstruct.Mapping;
910
import org.springframework.data.domain.Page;
1011
import org.springframework.data.domain.PageImpl;
1112

@@ -15,6 +16,14 @@
1516
@AnnotateWith(NullUnmarked.class)
1617
@NullMarked
1718
public interface EmailMapper {
19+
@Mapping(source = "content.subject", target = "subject")
20+
@Mapping(source = "content.fromAddress", target = "fromAddress")
21+
@Mapping(source = "content.fromName", target = "fromName")
22+
@Mapping(source = "content.replyToAddress", target = "replyToAddress")
23+
@Mapping(source = "content.replyToName", target = "replyToName")
24+
@Mapping(source = "content.recipients", target = "recipients")
25+
@Mapping(source = "content.textBody", target = "textBody")
26+
@Mapping(source = "content.htmlBody", target = "htmlBody")
1827
EmailDto toDto(Email model);
1928

2029
List<EmailDto> toDto(List<Email> model);

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

Lines changed: 9 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -69,6 +69,15 @@ updated_at timestamp with time zone default now() not null,
6969
7070
alter table email_service_emails add column if not exists reply_to_address text;
7171
alter table email_service_emails add column if not exists reply_to_name text;
72+
73+
alter table email_service_emails add column if not exists attempts int default 0 not null;
74+
alter table email_service_emails add column if not exists execution_start_time timestamp with time zone;
75+
alter table email_service_emails add column if not exists execution_end_time timestamp with time zone;
76+
77+
update email_service_emails
78+
set execution_end_time = sent_at
79+
where sent_at is not null
80+
and execution_end_time is null;
7281
"""
7382
//@formatter:on
7483
);

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

Lines changed: 103 additions & 57 deletions
Original file line numberDiff line numberDiff line change
@@ -9,21 +9,27 @@
99
import it.aboutbits.springboot.emailservice.lib.jpa.EmailRepository;
1010
import it.aboutbits.springboot.emailservice.lib.model.Email;
1111
import it.aboutbits.springboot.emailservice.lib.model.EmailAttachment;
12+
import it.aboutbits.springboot.emailservice.lib.model.EmailContent;
1213
import jakarta.mail.MessagingException;
1314
import jakarta.validation.Valid;
1415
import lombok.extern.slf4j.Slf4j;
1516
import org.jspecify.annotations.NullMarked;
1617
import org.jspecify.annotations.Nullable;
1718
import org.springframework.core.io.ByteArrayResource;
18-
import org.springframework.mail.MailException;
1919
import org.springframework.mail.javamail.JavaMailSender;
2020
import org.springframework.mail.javamail.MimeMessageHelper;
21+
import org.springframework.transaction.PlatformTransactionManager;
22+
import org.springframework.transaction.TransactionDefinition;
23+
import org.springframework.transaction.annotation.Transactional;
24+
import org.springframework.transaction.support.TransactionTemplate;
2125
import org.springframework.validation.annotation.Validated;
2226

2327
import java.io.IOException;
28+
import java.time.Duration;
2429
import java.time.OffsetDateTime;
2530
import java.util.HashSet;
2631
import java.util.List;
32+
import java.util.Optional;
2733
import java.util.Set;
2834

2935

@@ -35,17 +41,28 @@ public class ManageEmail {
3541
private final JavaMailSender mailSender;
3642
private final AttachmentDataSource attachmentDataSource;
3743
private final EmailMapper emailMapper;
44+
private final int maxAttempts;
45+
private final Duration schedulerInterval;
46+
private final TransactionTemplate transactionTemplate;
3847

3948
public ManageEmail(
4049
EmailRepository emailRepository,
4150
JavaMailSender mailSender,
4251
AttachmentDataSource attachmentDataSource,
43-
final EmailMapper emailMapper
52+
EmailMapper emailMapper,
53+
int maxAttempts,
54+
Duration schedulerInterval,
55+
PlatformTransactionManager transactionManager
4456
) {
4557
this.emailRepository = emailRepository;
4658
this.mailSender = mailSender;
4759
this.attachmentDataSource = attachmentDataSource;
4860
this.emailMapper = emailMapper;
61+
this.maxAttempts = maxAttempts;
62+
this.schedulerInterval = schedulerInterval;
63+
// Persist the final email state in its own, independent transaction for sendOrFail to never make it roll back
64+
this.transactionTemplate = new TransactionTemplate(transactionManager);
65+
this.transactionTemplate.setPropagationBehavior(TransactionDefinition.PROPAGATION_REQUIRES_NEW);
4966
}
5067

5168
public EmailDto schedule(@Valid EmailParameter parameter) throws EmailException {
@@ -69,33 +86,60 @@ public EmailDto sendOrFail(@Valid EmailParameter parameter) throws EmailExceptio
6986
throw new EmailException(e);
7087
}
7188

72-
var savedEmail = send(email);
89+
email.setExecutionStartTime(OffsetDateTime.now());
90+
email.incrementAttempts();
7391

74-
if (savedEmail.hasFailed()) {
92+
try {
93+
sendMail(email);
94+
email.setState(EmailState.SENT);
95+
email.setExecutionEndTime(OffsetDateTime.now());
96+
} catch (MessagingException | AttachmentException | IOException | RuntimeException e) {
97+
log.error("Failed to send email: {}", email.getId(), e);
98+
email.setState(EmailState.ERROR);
99+
email.setExecutionEndTime(OffsetDateTime.now());
100+
email.setErrorMessage(e.getMessage());
101+
}
102+
103+
var savedEmail = transactionTemplate.execute(_ -> emailRepository.save(email));
104+
105+
if (savedEmail.getState() == EmailState.ERROR) {
75106
throw new EmailException("Failed to send email [id=%s, providerMessage=%s]"
76107
.formatted(savedEmail.getId(), savedEmail.getErrorMessage()));
77108
}
78109

79110
return emailMapper.toDto(savedEmail);
80111
}
81112

82-
Email send(Email email) {
83-
if (EmailState.SENT.equals(email.getState())) {
84-
return email;
85-
}
113+
// Atomically transitions the row from PENDING/stale-SENDING into SENDING
114+
@Transactional
115+
Optional<Email> tryClaimForSend(long id, OffsetDateTime staleSendingBefore) {
116+
var claimed = emailRepository.claimForSend(id, OffsetDateTime.now(), staleSendingBefore);
117+
return claimed == 0 ? Optional.empty() : emailRepository.findById(id);
118+
}
86119

120+
// Actually try sending the claimed email
121+
Email completeClaimedSend(Email email) {
87122
try {
88123
sendMail(email);
89124
email.setState(EmailState.SENT);
90-
email.setErrorMessage("");
91-
email.setSentAt(OffsetDateTime.now());
92-
} catch (MailException | MessagingException | AttachmentException | IOException e) {
93-
log.error("Failed to send email: " + email.getId(), e);
125+
email.setExecutionEndTime(OffsetDateTime.now());
126+
email.setErrorMessage(null);
127+
} catch (Exception e) {
128+
email.setExecutionEndTime(OffsetDateTime.now());
94129
email.setErrorMessage(e.getMessage());
95-
email.setState(EmailState.ERROR);
130+
if (email.getAttempts() >= maxAttempts) {
131+
log.error("Failed to send email: {}", email.getId(), e);
132+
email.setState(EmailState.ERROR);
133+
} else {
134+
log.warn("Failed to send email: {}; Will be tried again", email.getId(), e);
135+
email.setState(EmailState.PENDING);
136+
email.setScheduledAt(
137+
OffsetDateTime.now()
138+
.plus(schedulerInterval.multipliedBy((long) Math.pow(2, email.getAttempts())))
139+
);
140+
}
96141
}
97-
98-
return emailRepository.save(email);
142+
return transactionTemplate.execute(_ -> emailRepository.save(email));
99143
}
100144

101145
void cleanupAttachments(final Email email) throws AttachmentException {
@@ -106,49 +150,17 @@ void cleanupAttachments(final Email email) throws AttachmentException {
106150
emailRepository.save(email);
107151
}
108152

109-
private Email fromParameter(EmailParameter parameter) throws AttachmentException {
110-
var emailData = parameter.email();
111-
112-
final var email = new Email();
113-
email.setState(EmailState.PENDING);
114-
email.setScheduledAt(parameter.scheduledAt());
115-
email.setSubject(emailData.subject());
116-
email.setTextBody(emailData.textBody());
117-
email.setHtmlBody(emailData.htmlBody());
118-
email.setRecipients(emailData.recipients());
119-
email.setFromAddress(emailData.fromAddress());
120-
email.setFromName(emailData.fromName());
121-
email.setReplyToAddress(emailData.replyToAddress());
122-
email.setReplyToName(emailData.replyToName());
123-
124-
var attachments = new HashSet<EmailAttachment>();
125-
for (var attachment : parameter.email().attachments()) {
126-
var reference = attachmentDataSource.storeAttachmentPayload(attachment.payload());
127-
128-
var emailAttachment = new EmailAttachment();
129-
emailAttachment.setEmail(email);
130-
emailAttachment.setContentType(attachment.contentType());
131-
emailAttachment.setFileName(attachment.fileName());
132-
emailAttachment.setFileReference(reference);
133-
134-
attachments.add(emailAttachment);
135-
}
136-
137-
email.setAttachments(attachments);
138-
139-
return email;
140-
}
141-
142153
private void sendMail(Email email) throws MessagingException, IOException, AttachmentException {
154+
var content = email.getContent();
143155
sendMail(
144-
email.getFromAddress(),
145-
email.getFromName(),
146-
email.getReplyToAddress(),
147-
email.getReplyToName(),
148-
email.getRecipients(),
149-
email.getSubject(),
150-
email.getHtmlBody(),
151-
email.getTextBody(),
156+
content.fromAddress(),
157+
content.fromName(),
158+
content.replyToAddress(),
159+
content.replyToName(),
160+
content.recipients(),
161+
content.subject(),
162+
content.htmlBody(),
163+
content.textBody(),
152164
email.getAttachments()
153165
);
154166
}
@@ -194,7 +206,41 @@ private void sendMail(
194206
payload.close();
195207
}
196208

197-
198209
mailSender.send(message);
199210
}
211+
212+
private Email fromParameter(EmailParameter parameter) throws AttachmentException {
213+
var emailData = parameter.email();
214+
215+
var email = new Email();
216+
email.setState(EmailState.PENDING);
217+
email.setScheduledAt(parameter.scheduledAt());
218+
email.setContent(new EmailContent(
219+
emailData.subject(),
220+
emailData.fromAddress(),
221+
emailData.fromName(),
222+
emailData.replyToAddress(),
223+
emailData.replyToName(),
224+
emailData.recipients(),
225+
emailData.textBody(),
226+
emailData.htmlBody()
227+
));
228+
229+
var attachments = new HashSet<EmailAttachment>();
230+
for (var attachment : parameter.email().attachments()) {
231+
var reference = attachmentDataSource.storeAttachmentPayload(attachment.payload());
232+
233+
var emailAttachment = new EmailAttachment();
234+
emailAttachment.setEmail(email);
235+
emailAttachment.setContentType(attachment.contentType());
236+
emailAttachment.setFileName(attachment.fileName());
237+
emailAttachment.setFileReference(reference);
238+
239+
attachments.add(emailAttachment);
240+
}
241+
242+
email.setAttachments(attachments);
243+
244+
return email;
245+
}
200246
}

0 commit comments

Comments
 (0)