Skip to content

Commit e0c13be

Browse files
jon-signaljon-signal
authored
Re-parallelize AbstractAttemptAnalyzer and InfobipSmsAttemptAnalyzer 
1 parent 87de2b1 commit e0c13be

11 files changed

Lines changed: 139 additions & 122 deletions

‎pom.xml‎

Lines changed: 13 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -57,6 +57,14 @@
5757
<scope>import</scope>
5858
</dependency>
5959

60+
<dependency>
61+
<groupId>io.projectreactor</groupId>
62+
<artifactId>reactor-bom</artifactId>
63+
<version>${reactor.bom.version}</version>
64+
<type>pom</type>
65+
<scope>import</scope>
66+
</dependency>
67+
6068
<dependency>
6169
<groupId>org.testcontainers</groupId>
6270
<artifactId>testcontainers-bom</artifactId>
@@ -281,6 +289,11 @@
281289
<artifactId>micronaut-validation</artifactId>
282290
</dependency>
283291

292+
<dependency>
293+
<groupId>io.projectreactor</groupId>
294+
<artifactId>reactor-core</artifactId>
295+
</dependency>
296+
284297
<dependency>
285298
<groupId>jakarta.annotation</groupId>
286299
<artifactId>jakarta.annotation-api</artifactId>

‎src/main/java/org/signal/registration/analytics/AbstractAttemptAnalyzer.java‎

Lines changed: 19 additions & 14 deletions
Original file line numberDiff line numberDiff line change
@@ -13,6 +13,9 @@
1313
import org.signal.registration.sender.VerificationCodeSender;
1414
import org.slf4j.Logger;
1515
import org.slf4j.LoggerFactory;
16+
import reactor.core.publisher.Flux;
17+
import reactor.core.publisher.Mono;
18+
import reactor.core.scheduler.Scheduler;
1619

1720
/**
1821
* An abstract attempt analyzer periodically reads attempts pending analysis from a
@@ -25,6 +28,7 @@
2528
public abstract class AbstractAttemptAnalyzer {
2629

2730
private final AttemptPendingAnalysisRepository repository;
31+
private final Scheduler scheduler;
2832
private final ApplicationEventPublisher<AttemptAnalyzedEvent> attemptAnalyzedEventPublisher;
2933
private final Clock clock;
3034

@@ -34,38 +38,39 @@ public abstract class AbstractAttemptAnalyzer {
3438
public static final Duration DEFAULT_PRICING_DEADLINE = Duration.ofHours(36);
3539

3640
protected AbstractAttemptAnalyzer(final AttemptPendingAnalysisRepository repository,
41+
final Scheduler scheduler,
3742
final ApplicationEventPublisher<AttemptAnalyzedEvent> attemptAnalyzedEventPublisher,
3843
final Clock clock) {
3944

4045
this.repository = repository;
46+
this.scheduler = scheduler;
4147
this.attemptAnalyzedEventPublisher = attemptAnalyzedEventPublisher;
4248
this.clock = clock;
4349
}
4450

4551
protected void analyzeAttempts() {
4652
logger.debug("Processing attempts pending analysis");
4753

48-
repository.getBySender(getSenderName())
49-
.map(attemptPendingAnalysis -> {
50-
try {
51-
return new AttemptAnalyzedEvent(attemptPendingAnalysis, analyzeAttempt(attemptPendingAnalysis));
52-
} catch (final RuntimeException e) {
53-
logger.warn("Failed to analyze attempt", e);
54-
return new AttemptAnalyzedEvent(attemptPendingAnalysis, AttemptAnalysis.EMPTY);
55-
}
56-
})
54+
Flux.fromStream(repository.getBySender(getSenderName()))
55+
.flatMap(attemptPendingAnalysis -> Mono.fromSupplier(() -> analyzeAttempt(attemptPendingAnalysis))
56+
.onErrorResume(throwable -> {
57+
logger.warn("Failed to analyze attempt pending analysis", throwable);
58+
return Mono.empty();
59+
})
60+
.subscribeOn(scheduler)
61+
.map(analysis -> new AttemptAnalyzedEvent(attemptPendingAnalysis, analysis)))
5762
.filter(attemptAnalyzedEvent -> {
58-
final Instant attemptTimestamp =
59-
Instant.ofEpochMilli(attemptAnalyzedEvent.attemptPendingAnalysis().getTimestampEpochMillis());
60-
63+
final Instant attemptTimestamp = Instant.ofEpochMilli(attemptAnalyzedEvent.attemptPendingAnalysis().getTimestampEpochMillis());
6164
final boolean pricingDeadlinePassed = clock.instant().isAfter(attemptTimestamp.plus(getPricingDeadline()));
6265

6366
return attemptAnalyzedEvent.attemptAnalysis().price().isPresent() || pricingDeadlinePassed;
6467
})
65-
.forEach(attemptAnalyzedEvent -> {
68+
.doOnNext(attemptAnalyzedEvent -> {
6669
repository.remove(attemptAnalyzedEvent.attemptPendingAnalysis());
6770
attemptAnalyzedEventPublisher.publishEvent(attemptAnalyzedEvent);
68-
});
71+
})
72+
.then()
73+
.block();
6974
}
7075

7176
/**

‎src/main/java/org/signal/registration/analytics/infobip/InfobipSmsAttemptAnalyzer.java‎

Lines changed: 60 additions & 101 deletions
Original file line numberDiff line numberDiff line change
@@ -18,13 +18,10 @@
1818
import java.time.Clock;
1919
import java.time.Duration;
2020
import java.time.Instant;
21-
import java.util.ArrayList;
2221
import java.util.Currency;
23-
import java.util.Iterator;
2422
import java.util.List;
2523
import java.util.Map;
2624
import java.util.Optional;
27-
import java.util.function.Function;
2825
import java.util.stream.Collectors;
2926
import javax.annotation.Nullable;
3027
import org.apache.arrow.util.VisibleForTesting;
@@ -38,6 +35,10 @@
3835
import org.signal.registration.sender.infobip.classic.InfobipSmsSender;
3936
import org.slf4j.Logger;
4037
import org.slf4j.LoggerFactory;
38+
import reactor.core.publisher.Flux;
39+
import reactor.core.publisher.Mono;
40+
import reactor.core.scheduler.Schedulers;
41+
import reactor.util.retry.Retry;
4142

4243
@Singleton
4344
class InfobipSmsAttemptAnalyzer {
@@ -50,10 +51,9 @@ class InfobipSmsAttemptAnalyzer {
5051
private final Currency defaultPriceCurrency;
5152
private final MeterRegistry meterRegistry;
5253
private final int pageSize;
53-
5454
private static final Logger logger = LoggerFactory.getLogger(InfobipSmsAttemptAnalyzer.class);
5555
private static final int MIN_MCC_MNC_LENGTH = 5;
56-
private static final int MAX_ATTEMPTS = 10;
56+
private static final int MAX_RETRIES = 10;
5757
private static final Duration MIN_BACKOFF = Duration.ofMillis(500);
5858
private static final Duration MAX_BACKOFF = Duration.ofSeconds(60);
5959
private static final Duration PRICING_DEADLINE = Duration.ofHours(36);
@@ -68,6 +68,7 @@ protected InfobipSmsAttemptAnalyzer(
6868
@Value("${analytics.infobip.sms.default-price-currency:USD}") final String defaultPriceCurrency,
6969
final MeterRegistry meterRegistry,
7070
@Value("${analytics.infobip.sms.page-size}") final int pageSize) {
71+
7172
this.repository = repository;
7273
this.attemptAnalyzedEventPublisher = attemptAnalyzedEventPublisher;
7374
this.clock = clock;
@@ -83,104 +84,62 @@ protected void analyzeAttempts() {
8384
// Infobip's message logs API (https://www.infobip.com/docs/api/channels/sms/sms-messaging/logs-and-status-reports/get-outbound-sms-message-logs)
8485
// has a ratelimit that prevents us from querying one by one for each attempt pending analysis.
8586
// Instead, we fetch fewer, larger pages and reconcile them against what we have stored locally.
86-
final List<AttemptPendingAnalysis> buffer = new ArrayList<>(pageSize);
87-
final Iterator<AttemptPendingAnalysis> attemptsPendingAnalysis =
88-
repository.getBySender(InfobipSmsSender.SENDER_NAME).iterator();
89-
90-
while (attemptsPendingAnalysis.hasNext()) {
91-
buffer.add(attemptsPendingAnalysis.next());
92-
93-
if (buffer.size() == pageSize) {
94-
try {
95-
analyzeAttempts(buffer);
96-
} catch (final ApiException e) {
97-
logger.warn("Failed to analyze attempts", e);
98-
} finally {
99-
buffer.clear();
100-
}
101-
}
102-
}
103-
104-
if (!buffer.isEmpty()) {
105-
try {
106-
analyzeAttempts(buffer);
107-
} catch (ApiException e) {
108-
logger.warn("Failed to analyze attempts", e);
109-
}
110-
}
111-
}
112-
113-
private void analyzeAttempts(final List<AttemptPendingAnalysis> attemptsPendingAnalysis) throws ApiException {
114-
final Map<String, SmsLog> smsLogsByRemoteId =
115-
fetchSmsLogsWithBackoff(attemptsPendingAnalysis.stream().map(AttemptPendingAnalysis::getRemoteId).toList());
116-
117-
for (final AttemptPendingAnalysis attemptPendingAnalysis : attemptsPendingAnalysis) {
118-
final AttemptAnalysis attemptAnalysis;
119-
120-
if (smsLogsByRemoteId.containsKey(attemptPendingAnalysis.getRemoteId())) {
121-
final SmsLog smsLog = smsLogsByRemoteId.get(attemptPendingAnalysis.getRemoteId());
122-
final MccMnc mccMnc = MccMnc.fromString(smsLog.getMccMnc());
123-
final Optional<Money> maybeEstimatedPriceByMccMnc = estimatePrice(mccMnc.toString());
124-
125-
attemptAnalysis = new AttemptAnalysis(
126-
extractPrice(smsLog),
127-
maybeEstimatedPriceByMccMnc.or(() -> estimatePrice(attemptPendingAnalysis.getRegion())),
128-
Optional.ofNullable(mccMnc.mcc),
129-
Optional.ofNullable(mccMnc.mnc));
130-
} else {
131-
attemptAnalysis = new AttemptAnalysis(
132-
Optional.empty(),
133-
estimatePrice(attemptPendingAnalysis.getRegion()),
134-
Optional.empty(),
135-
Optional.empty());
136-
}
137-
138-
final Instant attemptTimestamp = Instant.ofEpochMilli(attemptPendingAnalysis.getTimestampEpochMillis());
139-
final boolean pricingDeadlineExpired = clock.instant().isAfter(attemptTimestamp.plus(PRICING_DEADLINE));
140-
141-
if (attemptAnalysis.price().isPresent() || pricingDeadlineExpired) {
142-
meterRegistry.counter(MetricsUtil.name(getClass(), "attemptAnalyzed"),
143-
"hasPrice", String.valueOf(attemptAnalysis.price().isPresent())).increment();
144-
repository.remove(attemptPendingAnalysis);
145-
attemptAnalyzedEventPublisher.publishEvent(new AttemptAnalyzedEvent(attemptPendingAnalysis, attemptAnalysis));
146-
}
147-
}
87+
Flux.fromStream(repository.getBySender(InfobipSmsSender.SENDER_NAME))
88+
.buffer(pageSize)
89+
.flatMap(attemptsPendingAnalysis -> fetchSmsLogsWithBackoff(attemptsPendingAnalysis.stream().map(AttemptPendingAnalysis::getRemoteId).toList())
90+
.flatMapMany(smsLogsByRemoteId -> Flux.fromIterable(attemptsPendingAnalysis)
91+
.map(attemptPendingAnalysis -> {
92+
final AttemptAnalysis attemptAnalysis;
93+
94+
if (smsLogsByRemoteId.containsKey(attemptPendingAnalysis.getRemoteId())) {
95+
final SmsLog smsLog = smsLogsByRemoteId.get(attemptPendingAnalysis.getRemoteId());
96+
final MccMnc mccMnc = MccMnc.fromString(smsLog.getMccMnc());
97+
final Optional<Money> maybeEstimatedPriceByMccMnc = estimatePrice(mccMnc.toString());
98+
99+
attemptAnalysis = new AttemptAnalysis(
100+
extractPrice(smsLog),
101+
maybeEstimatedPriceByMccMnc.or(() -> estimatePrice(attemptPendingAnalysis.getRegion())),
102+
Optional.ofNullable(mccMnc.mcc),
103+
Optional.ofNullable(mccMnc.mnc));
104+
} else {
105+
attemptAnalysis = new AttemptAnalysis(
106+
Optional.empty(),
107+
estimatePrice(attemptPendingAnalysis.getRegion()),
108+
Optional.empty(),
109+
Optional.empty());
110+
}
111+
112+
return new AttemptAnalyzedEvent(attemptPendingAnalysis, attemptAnalysis);
113+
})), 1) // Request at most one page of results from Infobip at a time
114+
.filter(attemptAnalyzedEvent -> {
115+
final Instant attemptTimestamp = Instant.ofEpochMilli(attemptAnalyzedEvent.attemptPendingAnalysis().getTimestampEpochMillis());
116+
final boolean pricingDeadlineExpired = clock.instant().isAfter(attemptTimestamp.plus(PRICING_DEADLINE));
117+
118+
return attemptAnalyzedEvent.attemptAnalysis().price().isPresent() || pricingDeadlineExpired;
119+
})
120+
.doOnNext(attemptAnalyzedEvent -> {
121+
meterRegistry.counter(MetricsUtil.name(getClass(), "attemptAnalyzed"),
122+
"hasPrice", String.valueOf(attemptAnalyzedEvent.attemptAnalysis().price().isPresent())).increment();
123+
repository.remove(attemptAnalyzedEvent.attemptPendingAnalysis());
124+
attemptAnalyzedEventPublisher.publishEvent(attemptAnalyzedEvent);
125+
})
126+
.then()
127+
.block();
148128
}
149129

150-
private Map<String, SmsLog> fetchSmsLogsWithBackoff(final List<String> remoteIds) throws ApiException {
151-
Duration backoffDelay = MIN_BACKOFF;
152-
ApiException lastException = null;
153-
154-
for (int attempt = 0; attempt < MAX_ATTEMPTS; attempt++) {
155-
try {
156-
final List<SmsLog> smsLogs = getSmsLogs(remoteIds);
157-
return smsLogs.stream().collect(Collectors.toMap(SmsLog::getMessageId, Function.identity()));
158-
} catch (final ApiException e) {
159-
lastException = e;
160-
161-
if (e.responseStatusCode() == HTTP_TOO_MANY_REQUESTS_CODE) {
162-
try {
163-
Thread.sleep(backoffDelay);
164-
} catch (final InterruptedException ignored) {
165-
}
166-
167-
backoffDelay = backoffDelay.multipliedBy(2);
168-
169-
if (backoffDelay.compareTo(MAX_BACKOFF) > 0) {
170-
backoffDelay = MAX_BACKOFF;
171-
}
172-
} else {
173-
throw e;
174-
}
175-
}
176-
}
177-
178-
throw lastException;
130+
private Mono<Map<String, SmsLog>> fetchSmsLogsWithBackoff(final List<String> remoteIds) {
131+
return Mono.fromCallable(() -> getSmsLogs(remoteIds))
132+
.retryWhen(Retry.backoff(MAX_RETRIES, MIN_BACKOFF)
133+
.filter(throwable -> throwable instanceof ApiException apiException &&
134+
apiException.responseStatusCode() == HTTP_TOO_MANY_REQUESTS_CODE)
135+
.maxBackoff(MAX_BACKOFF))
136+
.map(smsLogs -> smsLogs.stream().collect(Collectors.toMap(SmsLog::getMessageId, smsLog -> smsLog)));
179137
}
180138

181139
private List<SmsLog> getSmsLogs(final List<String> remoteIds) throws ApiException {
182140
try {
183-
final List<SmsLog> smsLogs = infobipSmsApiClient.getOutboundSmsMessageLogs().messageId(remoteIds).execute().getResults();
141+
final List<SmsLog> smsLogs =
142+
infobipSmsApiClient.getOutboundSmsMessageLogs().messageId(remoteIds).execute().getResults();
184143
meterRegistry.counter(MetricsUtil.name(getClass(), "getSmsLogs")).increment(smsLogs.size());
185144
return smsLogs;
186145
} catch (final ApiException e) {
@@ -196,10 +155,10 @@ private List<SmsLog> getSmsLogs(final List<String> remoteIds) throws ApiExceptio
196155

197156
private Optional<Money> extractPrice(final SmsLog smsLog) {
198157
final boolean hasPriceData = smsLog.getPrice() != null && smsLog.getPrice().getPricePerMessage() != null
199-
&& smsLog.getPrice().getCurrency() != null;
158+
&& smsLog.getPrice().getCurrency() != null;
200159
return hasPriceData
201-
? Optional.of(new Money(BigDecimal.valueOf(smsLog.getPrice().getPricePerMessage()), Currency.getInstance(smsLog.getPrice().getCurrency())))
202-
: Optional.empty();
160+
? Optional.of(new Money(BigDecimal.valueOf(smsLog.getPrice().getPricePerMessage()), Currency.getInstance(smsLog.getPrice().getCurrency())))
161+
: Optional.empty();
203162
}
204163

205164
private Optional<Money> estimatePrice(final String key) {
@@ -225,7 +184,7 @@ static MccMnc fromString(@Nullable final String mccMnc) {
225184
// Mobile country code is always 3 digits: https://en.wikipedia.org/wiki/Mobile_country_code
226185
return new MccMnc(mccMnc.substring(0, 3), mccMnc.substring(3));
227186
}
228-
187+
229188
public String toString() {
230189
return mcc + mnc;
231190
}

‎src/main/java/org/signal/registration/analytics/messagebird/AbstractMessageBirdAttemptAnalyzer.java‎

Lines changed: 3 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -22,16 +22,18 @@
2222
import org.signal.registration.analytics.AttemptPendingAnalysisRepository;
2323
import org.signal.registration.analytics.Money;
2424
import org.signal.registration.analytics.PriceEstimator;
25+
import reactor.core.scheduler.Scheduler;
2526

2627
abstract class AbstractMessageBirdAttemptAnalyzer extends AbstractAttemptAnalyzer {
2728

2829
private final PriceEstimator priceEstimator;
2930

3031
AbstractMessageBirdAttemptAnalyzer(final AttemptPendingAnalysisRepository repository,
32+
final Scheduler scheduler,
3133
final ApplicationEventPublisher<AttemptAnalyzedEvent> attemptAnalyzedEventPublisher,
3234
final Clock clock, final PriceEstimator priceEstimator) {
3335

34-
super(repository, attemptAnalyzedEventPublisher, clock);
36+
super(repository, scheduler, attemptAnalyzedEventPublisher, clock);
3537

3638
this.priceEstimator = priceEstimator;
3739
}

‎src/main/java/org/signal/registration/analytics/messagebird/MessageBirdAnalysisSchedulerFactory.java‎

Lines changed: 28 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,28 @@
1+
/*
2+
* Copyright 2025 Signal Messenger, LLC
3+
* SPDX-License-Identifier: AGPL-3.0-only
4+
*/
5+
6+
package org.signal.registration.analytics.messagebird;
7+
8+
import io.micronaut.context.annotation.Factory;
9+
import io.micronaut.context.annotation.Property;
10+
import jakarta.inject.Named;
11+
import jakarta.inject.Singleton;
12+
import reactor.core.scheduler.Scheduler;
13+
import reactor.core.scheduler.Schedulers;
14+
15+
@Factory
16+
class MessageBirdAnalysisSchedulerFactory {
17+
18+
static final String SCHEDULER_NAME = "messagebird-analysis";
19+
20+
@Singleton
21+
@Named(SCHEDULER_NAME)
22+
Scheduler messagebirdAnalysisScheduler(
23+
@Property(name = "analytics.messagebird.max-concurrency", defaultValue = "32") final int maxConcurrency,
24+
@Property(name = "analytics.messagebird.max-queued-tasks", defaultValue = "65_536") final int maxQueuedTasks) {
25+
26+
return Schedulers.newBoundedElastic(maxConcurrency, maxQueuedTasks, SCHEDULER_NAME);
27+
}
28+
}

0 commit comments

Comments
 (0)