Skip to content

Commit 6fa769d

Browse files
J0nasMayrJ0nasMayr
committed
solve concurrency issue between pods using atomic compare-and-set; Add new configurables and make sendOrFail not retry
1 parent b25377f commit 6fa769d

5 files changed

Lines changed: 202 additions & 87 deletions

File tree

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

Lines changed: 97 additions & 55 deletions
Original file line numberDiff line numberDiff line change
@@ -9,21 +9,24 @@
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.annotation.Transactional;
2122
import org.springframework.validation.annotation.Validated;
2223

2324
import java.io.IOException;
25+
import java.time.Duration;
2426
import java.time.OffsetDateTime;
2527
import java.util.HashSet;
2628
import java.util.List;
29+
import java.util.Optional;
2730
import java.util.Set;
2831

2932

@@ -35,17 +38,23 @@ public class ManageEmail {
3538
private final JavaMailSender mailSender;
3639
private final AttachmentDataSource attachmentDataSource;
3740
private final EmailMapper emailMapper;
41+
private final int maxAttempts;
42+
private final Duration schedulerInterval;
3843

3944
public ManageEmail(
4045
EmailRepository emailRepository,
4146
JavaMailSender mailSender,
4247
AttachmentDataSource attachmentDataSource,
43-
final EmailMapper emailMapper
48+
EmailMapper emailMapper,
49+
int maxAttempts,
50+
Duration schedulerInterval
4451
) {
4552
this.emailRepository = emailRepository;
4653
this.mailSender = mailSender;
4754
this.attachmentDataSource = attachmentDataSource;
4855
this.emailMapper = emailMapper;
56+
this.maxAttempts = maxAttempts;
57+
this.schedulerInterval = schedulerInterval;
4958
}
5059

5160
public EmailDto schedule(@Valid EmailParameter parameter) throws EmailException {
@@ -69,32 +78,62 @@ public EmailDto sendOrFail(@Valid EmailParameter parameter) throws EmailExceptio
6978
throw new EmailException(e);
7079
}
7180

72-
var savedEmail = send(email);
81+
email.setState(EmailState.SENDING);
82+
email.setExecutionStartTime(OffsetDateTime.now());
83+
email.setExecutionEndTime(null);
84+
email.setErrorMessage(null);
85+
email.incrementAttempts();
86+
emailRepository.save(email);
7387

74-
if (savedEmail.hasFailed()) {
88+
try {
89+
sendMail(email);
90+
email.setState(EmailState.SENT);
91+
email.setExecutionEndTime(OffsetDateTime.now());
92+
} catch (MessagingException | AttachmentException | IOException | RuntimeException e) {
93+
log.error("Failed to send email: {}", email.getId(), e);
94+
email.setState(EmailState.ERROR);
95+
email.setExecutionEndTime(OffsetDateTime.now());
96+
email.setErrorMessage(e.getMessage());
97+
}
98+
99+
var savedEmail = emailRepository.save(email);
100+
101+
if (savedEmail.getState() == EmailState.ERROR) {
75102
throw new EmailException("Failed to send email [id=%s, providerMessage=%s]"
76103
.formatted(savedEmail.getId(), savedEmail.getErrorMessage()));
77104
}
78105

79106
return emailMapper.toDto(savedEmail);
80107
}
81108

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

116+
// Actually try sending the claimed email
117+
Email completeClaimedSend(Email email) {
87118
try {
88119
sendMail(email);
89120
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);
121+
email.setExecutionEndTime(OffsetDateTime.now());
122+
email.setErrorMessage(null);
123+
} catch (Exception e) {
124+
email.setExecutionEndTime(OffsetDateTime.now());
94125
email.setErrorMessage(e.getMessage());
95-
email.setState(EmailState.ERROR);
126+
if (email.getAttempts() > maxAttempts) {
127+
log.error("Failed to send email: {}", email.getId(), e);
128+
email.setState(EmailState.ERROR);
129+
} else {
130+
log.warn("Failed to send email: {}; Will be tried again", email.getId(), e);
131+
email.setState(EmailState.PENDING);
132+
email.setScheduledAt(
133+
OffsetDateTime.now().plus(schedulerInterval.multipliedBy(email.getAttempts() * 2L))
134+
);
135+
}
96136
}
97-
98137
return emailRepository.save(email);
99138
}
100139

@@ -106,49 +145,17 @@ void cleanupAttachments(final Email email) throws AttachmentException {
106145
emailRepository.save(email);
107146
}
108147

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-
142148
private void sendMail(Email email) throws MessagingException, IOException, AttachmentException {
149+
var content = email.getContent();
143150
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(),
151+
content.fromAddress(),
152+
content.fromName(),
153+
content.replyToAddress(),
154+
content.replyToName(),
155+
content.recipients(),
156+
content.subject(),
157+
content.htmlBody(),
158+
content.textBody(),
152159
email.getAttachments()
153160
);
154161
}
@@ -197,4 +204,39 @@ private void sendMail(
197204

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

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

Lines changed: 6 additions & 23 deletions
Original file line numberDiff line numberDiff line change
@@ -5,7 +5,6 @@
55
import it.aboutbits.springboot.emailservice.lib.EmailState;
66
import it.aboutbits.springboot.emailservice.lib.jpa.EmailRepository;
77
import it.aboutbits.springboot.emailservice.lib.model.Email;
8-
import jakarta.persistence.EntityManager;
98
import lombok.RequiredArgsConstructor;
109
import org.jspecify.annotations.NullMarked;
1110
import org.springframework.data.domain.Page;
@@ -23,7 +22,6 @@
2322
public class QueryEmail {
2423
private final EmailRepository emailRepository;
2524
private final EmailMapper emailMapper;
26-
private final EntityManager entityManager;
2725

2826
public Page<EmailDto> paginatedByState(EmailState state, PageRequest pageParameter) {
2927
var pageRequest = PageRequest.of(
@@ -42,30 +40,15 @@ public List<EmailDto> byIds(Collection<Long> ids) {
4240
return emailMapper.toDto(emailRepository.findByIdIn(ids));
4341
}
4442

45-
List<Email> readyToSend() {
46-
var entityGraph = entityManager.getEntityGraph("email_service_emails-entity-graph");
47-
return entityManager.createQuery(
48-
"""
49-
SELECT e from Email e WHERE e.scheduledAt < :scheduledBefore AND e.state IN (
50-
it.aboutbits.springboot.emailservice.lib.EmailState.PENDING,
51-
it.aboutbits.springboot.emailservice.lib.EmailState.ERROR
52-
)
53-
""", Email.class
54-
)
55-
.setParameter("scheduledBefore", OffsetDateTime.now())
56-
.setHint("jakarta.persistence.fetchgraph", entityGraph)
57-
.getResultList();
43+
List<Long> candidateIdsToSend(OffsetDateTime staleSendingBefore) {
44+
return emailRepository.findCandidateIdsToSend(
45+
OffsetDateTime.now(),
46+
staleSendingBefore
47+
);
5848
}
5949

6050
List<Email> readyToCleanup() {
61-
var entityGraph = entityManager.getEntityGraph("email_service_emails-entity-graph");
62-
return entityManager.createQuery(
63-
"""
64-
SELECT e from Email e WHERE e.attachmentsCleaned=false AND e.state=it.aboutbits.springboot.emailservice.lib.EmailState.SENT
65-
""", Email.class
66-
)
67-
.setHint("jakarta.persistence.fetchgraph", entityGraph)
68-
.getResultList();
51+
return emailRepository.findReadyToCleanup();
6952
}
7053

7154
public Optional<EmailDto> byId(long id) {

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

Lines changed: 27 additions & 9 deletions
Original file line numberDiff line numberDiff line change
@@ -2,15 +2,14 @@
22

33

44
import it.aboutbits.springboot.emailservice.lib.EmailSchedulerCallback;
5-
import lombok.RequiredArgsConstructor;
65
import lombok.extern.log4j.Log4j2;
76
import org.jspecify.annotations.NullMarked;
87
import org.springframework.scheduling.annotation.Scheduled;
98

109
import java.time.Duration;
10+
import java.time.OffsetDateTime;
1111
import java.util.List;
1212

13-
@RequiredArgsConstructor
1413
@Log4j2
1514
@NullMarked
1615
public class SendScheduledEmails {
@@ -19,27 +18,46 @@ public class SendScheduledEmails {
1918
private final QueryEmail queryEmail;
2019
private final ManageEmail manageEmail;
2120
private final List<EmailSchedulerCallback> callbacks;
21+
private final Duration stuckSendingRecoveryThreshold;
2222

2323
private long lastInfoLogMillis = System.currentTimeMillis();
2424
private long silentRuns = 0;
2525
private boolean firstRun = true;
2626

27+
public SendScheduledEmails(
28+
QueryEmail queryEmail,
29+
ManageEmail manageEmail,
30+
List<EmailSchedulerCallback> callbacks,
31+
Duration stuckSendingRecoveryThreshold
32+
) {
33+
this.queryEmail = queryEmail;
34+
this.manageEmail = manageEmail;
35+
this.callbacks = callbacks;
36+
this.stuckSendingRecoveryThreshold = stuckSendingRecoveryThreshold;
37+
}
38+
2739
@Scheduled(initialDelayString = "${aboutbits.emailservice.scheduling.interval:30000}", fixedDelayString = "${aboutbits.emailservice.scheduling.interval:30000}")
2840
void sendEmails() {
2941
logStartOfPass();
3042

31-
var emailsToSend = queryEmail.readyToSend();
43+
var staleSendingBefore = OffsetDateTime.now().minus(stuckSendingRecoveryThreshold);
44+
var candidateIds = queryEmail.candidateIdsToSend(staleSendingBefore);
3245

3346
var countSent = 0;
3447
var countError = 0;
35-
for (var email : emailsToSend) {
36-
var updatedEmail = manageEmail.send(email);
37-
switch (updatedEmail.getState()) {
38-
case ERROR -> countError++;
48+
for (var id : candidateIds) {
49+
var claimed = manageEmail.tryClaimForSend(id, staleSendingBefore);
50+
if (claimed.isEmpty()) {
51+
// Lost race to another pod; Skip
52+
continue;
53+
}
54+
var updated = manageEmail.completeClaimedSend(claimed.get());
55+
switch (updated.getState()) {
56+
case ERROR, PENDING -> countError++;
3957
case SENT -> countSent++;
4058
default -> log.warn(
4159
JOB_DESCRIPTION + " | Job produced an invalid notification result state: {}.",
42-
updatedEmail.getState().name()
60+
updated.getState().name()
4361
);
4462
}
4563
}
@@ -48,7 +66,7 @@ void sendEmails() {
4866

4967
for (var callback : callbacks) {
5068
callback.report(new EmailSchedulerCallback.Report(
51-
emailsToSend.size(),
69+
candidateIds.size(),
5270
countSent,
5371
countError
5472
));

0 commit comments

Comments
 (0)