Skip to content

Commit 15c22d4

Browse files
J0nasMayrJ0nasMayr
committed
first draft for concurrency using an approach at the database level
1 parent 8ef5819 commit 15c22d4

10 files changed

Lines changed: 478 additions & 65 deletions

File tree

‎readme.md‎

Lines changed: 19 additions & 5 deletions
Original file line numberDiff line numberDiff line change
@@ -74,11 +74,25 @@ 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 | Specifies the milliseconds delay between runs of the scheduler. |
83+
| `aboutbits.emailservice.scheduling.batch-size` | 50 | Maximum number of emails a single pod claims per scheduler pass (see [Multi-pod deployments](#multi-pod-deployments)). |
84+
85+
## Multi-pod deployments
86+
87+
The scheduler is safe to run in every pod concurrently. Each ready email is
88+
claimed by exactly one pod using Postgres row-level `SELECT ... FOR UPDATE SKIP LOCKED`,
89+
so pods work on disjoint rows in parallel and no email is ever sent twice by
90+
different pods. The same guarantee applies to the attachment cleanup scheduler.
91+
92+
`aboutbits.emailservice.scheduling.batch-size` caps the number of emails one
93+
pod processes per pass. With the default 30s interval and 50 emails per pass,
94+
a single pod can take up to 100 emails per minute; For higher-throughput deployments increase the batch size or
95+
lower the interval.
8296

8397
## Local development:
8498

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

Lines changed: 26 additions & 6 deletions
Original file line numberDiff line numberDiff line change
@@ -16,6 +16,7 @@
1616
import it.aboutbits.springboot.emailservice.lib.jpa.EmailRepository;
1717
import jakarta.persistence.EntityManager;
1818
import org.jspecify.annotations.NullMarked;
19+
import org.springframework.beans.factory.annotation.Value;
1920
import org.springframework.boot.autoconfigure.AutoConfigurationPackage;
2021
import org.springframework.boot.autoconfigure.condition.ConditionalOnMissingBean;
2122
import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
@@ -45,25 +46,44 @@ public EmailMapper emailMapper() {
4546
}
4647

4748
@Bean
48-
public QueryEmail queryEmail(EmailRepository emailRepository, EmailMapper emailMapper, EntityManager entityManager) {
49+
public QueryEmail queryEmail(
50+
EmailRepository emailRepository,
51+
EmailMapper emailMapper,
52+
EntityManager entityManager
53+
) {
4954
return new QueryEmail(emailRepository, emailMapper, entityManager);
5055
}
5156

5257
@Bean
53-
public ManageEmail manageEmail(EmailRepository emailRepository, JavaMailSender javaMailSender, AttachmentDataSource attachmentDataSource, EmailMapper emailMapper) {
58+
public ManageEmail manageEmail(
59+
EmailRepository emailRepository,
60+
JavaMailSender javaMailSender,
61+
AttachmentDataSource attachmentDataSource,
62+
EmailMapper emailMapper
63+
) {
5464
return new ManageEmail(emailRepository, javaMailSender, attachmentDataSource, emailMapper);
5565
}
5666

5767
@Bean
5868
@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);
69+
public SendScheduledEmails sendScheduledEmails(
70+
QueryEmail queryEmail,
71+
ManageEmail manageEmail,
72+
List<EmailSchedulerCallback> callbacks,
73+
@Value("${aboutbits.emailservice.scheduling.batch-size:50}") int batchSize
74+
) {
75+
return new SendScheduledEmails(queryEmail, manageEmail, callbacks, batchSize);
6176
}
6277

6378
@Bean
6479
@ConditionalOnProperty(value = "aboutbits.emailservice.scheduling.cleanup.enabled", matchIfMissing = true)
65-
public CleanupAttachmentFiles cleanupAttachments(QueryEmail queryEmail, ManageEmail manageEmail, List<AttachmentCleanerCallback> callbacks) {
66-
return new CleanupAttachmentFiles(queryEmail, manageEmail, callbacks);
80+
public CleanupAttachmentFiles cleanupAttachments(
81+
QueryEmail queryEmail,
82+
ManageEmail manageEmail,
83+
List<AttachmentCleanerCallback> callbacks,
84+
@Value("${aboutbits.emailservice.scheduling.batch-size:50}") int batchSize
85+
) {
86+
return new CleanupAttachmentFiles(queryEmail, manageEmail, callbacks, batchSize);
6787
}
6888

6989
@Bean

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

Lines changed: 22 additions & 20 deletions
Original file line numberDiff line numberDiff line change
@@ -2,16 +2,14 @@
22

33

44
import it.aboutbits.springboot.emailservice.lib.AttachmentCleanerCallback;
5-
import it.aboutbits.springboot.emailservice.lib.exception.AttachmentException;
6-
import lombok.RequiredArgsConstructor;
75
import lombok.extern.log4j.Log4j2;
86
import org.jspecify.annotations.NullMarked;
97
import org.springframework.scheduling.annotation.Scheduled;
8+
import org.springframework.transaction.annotation.Transactional;
109

1110
import java.time.Duration;
1211
import java.util.List;
1312

14-
@RequiredArgsConstructor
1513
@Log4j2
1614
@NullMarked
1715
public class CleanupAttachmentFiles {
@@ -20,35 +18,39 @@ public class CleanupAttachmentFiles {
2018
private final QueryEmail queryEmail;
2119
private final ManageEmail manageEmail;
2220
private final List<AttachmentCleanerCallback> callbacks;
21+
private final int batchSize;
2322

2423
private long lastInfoLogMillis = System.currentTimeMillis();
2524
private long silentRuns = 0;
2625
private boolean firstRun = true;
2726

27+
public CleanupAttachmentFiles(
28+
QueryEmail queryEmail,
29+
ManageEmail manageEmail,
30+
List<AttachmentCleanerCallback> callbacks,
31+
int batchSize
32+
) {
33+
this.queryEmail = queryEmail;
34+
this.manageEmail = manageEmail;
35+
this.callbacks = callbacks;
36+
this.batchSize = batchSize;
37+
}
38+
2839
@Scheduled(initialDelayString = "${aboutbits.emailservice.scheduling.interval:30000}", fixedDelayString = "${aboutbits.emailservice.scheduling.interval:30000}")
29-
void cleanupAttachments() {
40+
@Transactional
41+
void claimAndCleanupAttachments() {
3042
logStartOfPass();
3143

32-
var emailsToCleanup = queryEmail.readyToCleanup();
33-
34-
var countCleaned = 0;
35-
var countError = 0;
36-
for (var email : emailsToCleanup) {
37-
try {
38-
manageEmail.cleanupAttachments(email);
39-
countCleaned++;
40-
} catch (AttachmentException e) {
41-
countError++;
42-
}
43-
}
44+
var ids = queryEmail.claimReadyToCleanupIds(batchSize);
45+
var outcome = manageEmail.cleanupBatch(ids);
4446

45-
logEndOfPass(countCleaned, countError);
47+
logEndOfPass(outcome.cleaned(), outcome.errors());
4648

4749
for (var callback : callbacks) {
4850
callback.report(new AttachmentCleanerCallback.Report(
49-
emailsToCleanup.size(),
50-
countCleaned,
51-
countError
51+
outcome.total(),
52+
outcome.cleaned(),
53+
outcome.errors()
5254
));
5355
}
5456
}

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

Lines changed: 62 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -18,6 +18,7 @@
1818
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;
@@ -31,6 +32,18 @@
3132
@Slf4j
3233
@NullMarked
3334
public class ManageEmail {
35+
record SendBatchOutcome(int sent, int errors) {
36+
int total() {
37+
return sent + errors;
38+
}
39+
}
40+
41+
record CleanupBatchOutcome(int cleaned, int errors) {
42+
int total() {
43+
return cleaned + errors;
44+
}
45+
}
46+
3447
private final EmailRepository emailRepository;
3548
private final JavaMailSender mailSender;
3649
private final AttachmentDataSource attachmentDataSource;
@@ -40,7 +53,7 @@ public ManageEmail(
4053
EmailRepository emailRepository,
4154
JavaMailSender mailSender,
4255
AttachmentDataSource attachmentDataSource,
43-
final EmailMapper emailMapper
56+
EmailMapper emailMapper
4457
) {
4558
this.emailRepository = emailRepository;
4659
this.mailSender = mailSender;
@@ -79,6 +92,54 @@ public EmailDto sendOrFail(@Valid EmailParameter parameter) throws EmailExceptio
7992
return emailMapper.toDto(savedEmail);
8093
}
8194

95+
// Sends the emails identified by the given ids and persists SENT/ERROR state for each
96+
@Transactional
97+
SendBatchOutcome sendBatch(List<Long> ids) {
98+
if (ids.isEmpty()) {
99+
return new SendBatchOutcome(0, 0);
100+
}
101+
var emails = emailRepository.findByIdIn(ids);
102+
var sent = 0;
103+
var errors = 0;
104+
for (var email : emails) {
105+
try {
106+
var updated = send(email);
107+
if (updated.hasFailed()) {
108+
errors++;
109+
} else {
110+
sent++;
111+
}
112+
} catch (RuntimeException e) {
113+
// A single misbehaving email should not roll back the whole batch's committed state
114+
// (which would risk duplicate delivery for siblings that already left the SMTP relay).
115+
log.error("Unexpected failure while sending email: {}", email.getId(), e);
116+
errors++;
117+
}
118+
}
119+
return new SendBatchOutcome(sent, errors);
120+
}
121+
122+
// Releases attachment payloads for the given emails and marks them cleaned
123+
@Transactional
124+
CleanupBatchOutcome cleanupBatch(List<Long> ids) {
125+
if (ids.isEmpty()) {
126+
return new CleanupBatchOutcome(0, 0);
127+
}
128+
var emails = emailRepository.findByIdIn(ids);
129+
var cleaned = 0;
130+
var errors = 0;
131+
for (var email : emails) {
132+
try {
133+
cleanupAttachments(email);
134+
cleaned++;
135+
} catch (AttachmentException | RuntimeException e) {
136+
log.warn("Failed to cleanup attachments for email: {}", email.getId(), e);
137+
errors++;
138+
}
139+
}
140+
return new CleanupBatchOutcome(cleaned, errors);
141+
}
142+
82143
Email send(Email email) {
83144
if (EmailState.SENT.equals(email.getState())) {
84145
return email;

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

Lines changed: 19 additions & 11 deletions
Original file line numberDiff line numberDiff line change
@@ -4,8 +4,8 @@
44
import it.aboutbits.springboot.emailservice.lib.EmailDto;
55
import it.aboutbits.springboot.emailservice.lib.EmailState;
66
import it.aboutbits.springboot.emailservice.lib.jpa.EmailRepository;
7-
import it.aboutbits.springboot.emailservice.lib.model.Email;
87
import jakarta.persistence.EntityManager;
8+
import jakarta.persistence.LockModeType;
99
import lombok.RequiredArgsConstructor;
1010
import org.jspecify.annotations.NullMarked;
1111
import org.springframework.data.domain.Page;
@@ -21,6 +21,10 @@
2121
@RequiredArgsConstructor
2222
@NullMarked
2323
public class QueryEmail {
24+
// Hibernate convention for "jakarta.persistence.lock.timeout":
25+
// -2 translates to "SKIP LOCKED" at the database layer (matches "org.hibernate.Timeouts.SKIP_LOCKED_MILLI")
26+
private static final int SKIP_LOCKED_TIMEOUT = -2;
27+
private static final String LOCK_TIMEOUT_HINT = "jakarta.persistence.lock.timeout";
2428
private final EmailRepository emailRepository;
2529
private final EmailMapper emailMapper;
2630
private final EntityManager entityManager;
@@ -42,29 +46,33 @@ public List<EmailDto> byIds(Collection<Long> ids) {
4246
return emailMapper.toDto(emailRepository.findByIdIn(ids));
4347
}
4448

45-
List<Email> readyToSend() {
46-
var entityGraph = entityManager.getEntityGraph("email_service_emails-entity-graph");
49+
List<Long> claimReadyToSendIds(int limit) {
4750
return entityManager.createQuery(
4851
"""
49-
SELECT e from Email e WHERE e.scheduledAt < :scheduledBefore AND e.state IN (
52+
select e.id from Email e where e.scheduledAt < :scheduledBefore and e.state in (
5053
it.aboutbits.springboot.emailservice.lib.EmailState.PENDING,
5154
it.aboutbits.springboot.emailservice.lib.EmailState.ERROR
5255
)
53-
""", Email.class
56+
order by e.scheduledAt
57+
""", Long.class
5458
)
5559
.setParameter("scheduledBefore", OffsetDateTime.now())
56-
.setHint("jakarta.persistence.fetchgraph", entityGraph)
60+
.setLockMode(LockModeType.PESSIMISTIC_WRITE)
61+
.setHint(LOCK_TIMEOUT_HINT, SKIP_LOCKED_TIMEOUT)
62+
.setMaxResults(limit)
5763
.getResultList();
5864
}
5965

60-
List<Email> readyToCleanup() {
61-
var entityGraph = entityManager.getEntityGraph("email_service_emails-entity-graph");
66+
List<Long> claimReadyToCleanupIds(int limit) {
6267
return entityManager.createQuery(
6368
"""
64-
SELECT e from Email e WHERE e.attachmentsCleaned=false AND e.state=it.aboutbits.springboot.emailservice.lib.EmailState.SENT
65-
""", Email.class
69+
select e.id from Email e where e.attachmentsCleaned=false and e.state=it.aboutbits.springboot.emailservice.lib.EmailState.SENT
70+
order by e.updatedAt
71+
""", Long.class
6672
)
67-
.setHint("jakarta.persistence.fetchgraph", entityGraph)
73+
.setLockMode(LockModeType.PESSIMISTIC_WRITE)
74+
.setHint(LOCK_TIMEOUT_HINT, SKIP_LOCKED_TIMEOUT)
75+
.setMaxResults(limit)
6876
.getResultList();
6977
}
7078

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

Lines changed: 22 additions & 22 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;
8+
import org.springframework.transaction.annotation.Transactional;
99

1010
import java.time.Duration;
1111
import java.util.List;
1212

13-
@RequiredArgsConstructor
1413
@Log4j2
1514
@NullMarked
1615
public class SendScheduledEmails {
@@ -19,38 +18,39 @@ public class SendScheduledEmails {
1918
private final QueryEmail queryEmail;
2019
private final ManageEmail manageEmail;
2120
private final List<EmailSchedulerCallback> callbacks;
21+
private final int batchSize;
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+
int batchSize
32+
) {
33+
this.queryEmail = queryEmail;
34+
this.manageEmail = manageEmail;
35+
this.callbacks = callbacks;
36+
this.batchSize = batchSize;
37+
}
38+
2739
@Scheduled(initialDelayString = "${aboutbits.emailservice.scheduling.interval:30000}", fixedDelayString = "${aboutbits.emailservice.scheduling.interval:30000}")
28-
void sendEmails() {
40+
@Transactional
41+
void claimAndSendEmails() {
2942
logStartOfPass();
3043

31-
var emailsToSend = queryEmail.readyToSend();
32-
33-
var countSent = 0;
34-
var countError = 0;
35-
for (var email : emailsToSend) {
36-
var updatedEmail = manageEmail.send(email);
37-
switch (updatedEmail.getState()) {
38-
case ERROR -> countError++;
39-
case SENT -> countSent++;
40-
default -> log.warn(
41-
JOB_DESCRIPTION + " | Job produced an invalid notification result state: {}.",
42-
updatedEmail.getState().name()
43-
);
44-
}
45-
}
44+
var ids = queryEmail.claimReadyToSendIds(batchSize);
45+
var outcome = manageEmail.sendBatch(ids);
4646

47-
logEndOfPass(countSent, countError);
47+
logEndOfPass(outcome.sent(), outcome.errors());
4848

4949
for (var callback : callbacks) {
5050
callback.report(new EmailSchedulerCallback.Report(
51-
emailsToSend.size(),
52-
countSent,
53-
countError
51+
outcome.total(),
52+
outcome.sent(),
53+
outcome.errors()
5454
));
5555
}
5656
}

0 commit comments

Comments
 (0)