diff --git a/.gitignore b/.gitignore index bbdab3c..fbc89d9 100644 --- a/.gitignore +++ b/.gitignore @@ -85,3 +85,11 @@ application-secrets.yaml *.jks *.keystore secrets/ + +# --- Local dev tooling, not part of the service --- +# The Flutter desktop console: a developer client for this API, with its own +# toolchain and build output. Useful locally, but it is not the service and +# would drag a second SDK into anyone who clones this repo. +payment-console/ +# Throwaway experiments — provider mocks, load-test scripts, captured payloads. +scratchpad/ diff --git a/src/main/java/com/test/payment/client/AirtelClient.java b/src/main/java/com/test/payment/client/AirtelClient.java index 2ae9b2a..402059e 100644 --- a/src/main/java/com/test/payment/client/AirtelClient.java +++ b/src/main/java/com/test/payment/client/AirtelClient.java @@ -2,7 +2,6 @@ package com.test.payment.client; import com.test.payment.dto.AirtelPaymentRequestDto; import com.test.payment.dto.AirtelResponseDto; -import com.test.payment.exceptions.ProviderBusyException; import com.test.payment.service.AirtelTokenService; import io.github.resilience4j.circuitbreaker.annotation.CircuitBreaker; import io.github.resilience4j.ratelimiter.annotation.RateLimiter; @@ -15,7 +14,6 @@ import org.springframework.stereotype.Component; import org.springframework.web.reactive.function.client.WebClient; import reactor.core.publisher.Mono; -import java.time.Duration; @Component @RequiredArgsConstructor @@ -43,11 +41,7 @@ public class AirtelClient { .onStatus(HttpStatusCode::isError, resp -> ProviderHttpErrors.map(resp, "AIRTEL", tokenService.evictToken())) .bodyToMono(AirtelResponseDto.class)) - .retryWhen(reactor.util.retry.Retry.backoff(3, Duration.ofSeconds(3)) - .maxBackoff(Duration.ofSeconds(30)) - .jitter(0.5) - .filter(ProviderBusyException.class::isInstance) - .onRetryExhaustedThrow((spec, signal) -> signal.failure())); + .retryWhen(ProviderBackoff.whenBusy("AIRTEL")); } @CircuitBreaker(name = "airtelCircuitBreaker") diff --git a/src/main/java/com/test/payment/client/MpesaClient.java b/src/main/java/com/test/payment/client/MpesaClient.java index 933a145..2b59894 100644 --- a/src/main/java/com/test/payment/client/MpesaClient.java +++ b/src/main/java/com/test/payment/client/MpesaClient.java @@ -4,7 +4,6 @@ import com.test.payment.dto.MpesaRequestDto; import com.test.payment.dto.MpesaResponse; import com.test.payment.dto.StkQueryRequestDto; import com.test.payment.dto.StkQueryResponseDto; -import com.test.payment.exceptions.ProviderBusyException; import com.test.payment.exceptions.ProviderProcessingException; import com.test.payment.service.MpesaTokenService; import io.github.resilience4j.circuitbreaker.annotation.CircuitBreaker; @@ -18,15 +17,14 @@ import org.springframework.web.reactive.function.client.ClientResponse; import org.springframework.web.reactive.function.client.WebClient; import reactor.core.publisher.Mono; -import java.time.Duration; /** * All outbound Safaricom calls live here so the Resilience4j annotations are applied * through Spring AOP (they would be silently skipped on self-invocation inside a service). * - * The Resilience4j @Retry handles network-level failures; the reactive backoff retry - * below handles ProviderBusyException (rate limiting / "system busy"). Both coexist - * intentionally. + * The Resilience4j @Retry handles network-level failures; ProviderBackoff.whenBusy + * handles ProviderBusyException (rate limiting / "system busy") with exponential + * backoff. Both coexist intentionally. */ @Component @RequiredArgsConstructor @@ -48,11 +46,7 @@ public class MpesaClient { .retrieve() .onStatus(HttpStatusCode::isError, resp -> mapError(resp, false)) .bodyToMono(MpesaResponse.class)) - .retryWhen(reactor.util.retry.Retry.backoff(3, Duration.ofSeconds(3)) - .maxBackoff(Duration.ofSeconds(30)) - .jitter(0.5) - .filter(ProviderBusyException.class::isInstance) - .onRetryExhaustedThrow((spec, signal) -> signal.failure())); + .retryWhen(ProviderBackoff.whenBusy("MPESA")); } @CircuitBreaker(name = "mpesaCircuitBreaker") diff --git a/src/main/java/com/test/payment/client/MtnClient.java b/src/main/java/com/test/payment/client/MtnClient.java index 117186f..4fb7159 100644 --- a/src/main/java/com/test/payment/client/MtnClient.java +++ b/src/main/java/com/test/payment/client/MtnClient.java @@ -2,7 +2,6 @@ package com.test.payment.client; import com.test.payment.dto.MtnPayRequestDto; import com.test.payment.dto.MtnStatusResponseDto; -import com.test.payment.exceptions.ProviderBusyException; import com.test.payment.service.MtnTokenService; import io.github.resilience4j.circuitbreaker.annotation.CircuitBreaker; import io.github.resilience4j.ratelimiter.annotation.RateLimiter; @@ -16,7 +15,6 @@ import org.springframework.stereotype.Component; import org.springframework.web.reactive.function.client.WebClient; import reactor.core.publisher.Mono; -import java.time.Duration; @Component @RequiredArgsConstructor @@ -52,11 +50,7 @@ public class MtnClient { resp -> ProviderHttpErrors.map(resp, "MTN", tokenService.evictToken())) .toBodilessEntity() .then()) - .retryWhen(reactor.util.retry.Retry.backoff(3, Duration.ofSeconds(3)) - .maxBackoff(Duration.ofSeconds(30)) - .jitter(0.5) - .filter(ProviderBusyException.class::isInstance) - .onRetryExhaustedThrow((spec, signal) -> signal.failure())); + .retryWhen(ProviderBackoff.whenBusy("MTN")); } @CircuitBreaker(name = "mtnCircuitBreaker") diff --git a/src/main/java/com/test/payment/client/ProviderBackoff.java b/src/main/java/com/test/payment/client/ProviderBackoff.java new file mode 100644 index 0000000..1399115 --- /dev/null +++ b/src/main/java/com/test/payment/client/ProviderBackoff.java @@ -0,0 +1,65 @@ +package com.test.payment.client; + +import com.test.payment.exceptions.ProviderBusyException; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; +import reactor.util.retry.Retry; +import reactor.util.retry.RetryBackoffSpec; + +import java.time.Duration; + +/** + * The reactive backoff every provider client uses for {@link ProviderBusyException}. + * + *

This is a second, separate retry from the Resilience4j + * {@code @Retry} annotation, and the two coexist on purpose: {@code @Retry} handles + * network-level failures on a fixed 2s interval, while this one handles a provider + * that answered perfectly well to say "slow down" (429, or a busy body). Backing off + * a rate limit at a fixed interval just gets refused again, so this one is + * exponential with jitter — spreading concurrent callers out instead of having them + * all come back in lockstep. + * + *

It logs through the same {@code com.test.payment.resilience} logger as the + * Resilience4j event consumers, so one level setting covers every wait in the stack. + */ +public final class ProviderBackoff { + + private static final Logger log = LoggerFactory.getLogger("com.test.payment.resilience"); + + private static final int MAX_ATTEMPTS = 3; + private static final Duration FIRST_BACKOFF = Duration.ofSeconds(3); + private static final Duration MAX_BACKOFF = Duration.ofSeconds(30); + private static final double JITTER = 0.5; + + private ProviderBackoff() { + } + + /** + * @param provider operator name, for the log line — the spec is otherwise identical + * for every provider + */ + public static RetryBackoffSpec whenBusy(String provider) { + return Retry.backoff(MAX_ATTEMPTS, FIRST_BACKOFF) + .maxBackoff(MAX_BACKOFF) + .jitter(JITTER) + .filter(ProviderBusyException.class::isInstance) + // Reactor computes the delay internally and does not expose it to the + // callback, so the line reports the attempt rather than inventing a + // number that might not match what it actually sleeps. + .doBeforeRetry(signal -> log.warn( + "[{}] provider busy — backing off before retry {} of {} (exponential from {}s, " + + "capped at {}s, {}% jitter): {}", + provider, + signal.totalRetries() + 1, + MAX_ATTEMPTS, + FIRST_BACKOFF.toSeconds(), + MAX_BACKOFF.toSeconds(), + (int) (JITTER * 100), + signal.failure().getMessage())) + .onRetryExhaustedThrow((spec, signal) -> { + log.error("[{}] provider still busy after {} backoff attempt(s) — giving up", + provider, signal.totalRetries()); + return signal.failure(); + }); + } +} diff --git a/src/main/java/com/test/payment/configurations/ResilienceEventLogging.java b/src/main/java/com/test/payment/configurations/ResilienceEventLogging.java new file mode 100644 index 0000000..9cd69b3 --- /dev/null +++ b/src/main/java/com/test/payment/configurations/ResilienceEventLogging.java @@ -0,0 +1,237 @@ +package com.test.payment.configurations; + +import io.github.resilience4j.circuitbreaker.CircuitBreaker; +import io.github.resilience4j.core.registry.EntryAddedEvent; +import io.github.resilience4j.core.registry.EntryRemovedEvent; +import io.github.resilience4j.core.registry.EntryReplacedEvent; +import io.github.resilience4j.core.registry.RegistryEventConsumer; +import io.github.resilience4j.ratelimiter.RateLimiter; +import io.github.resilience4j.retry.Retry; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; +import org.springframework.context.annotation.Bean; +import org.springframework.context.annotation.Configuration; + +/** + * Makes the resilience layer audible. + * + *

Resilience4j publishes events but ships no logging of its own, so + * without these consumers a retry, an opening circuit breaker or a refused rate + * limiter permit is completely invisible: the call just takes longer, or fails + * with an exception that says nothing about which guard produced it. + * + *

Subscription happens at the registry level rather than per + * instance at startup, because Resilience4j creates instances lazily on first use + * — an instance that has never been called does not exist yet, so iterating the + * registry during {@code @PostConstruct} would find nothing to subscribe to. + * + *

Levels are chosen by operational meaning, not by event type: anything that + * says the provider is degrading is WARN, anything that means calls are actively + * being rejected is ERROR, recovery is INFO, and the per-call chatter is DEBUG. + * Everything logs through one logger name so it can be turned up or down in one + * place — {@code logging.level.com.test.payment.resilience}. + */ +@Configuration +public class ResilienceEventLogging { + + private static final Logger log = LoggerFactory.getLogger("com.test.payment.resilience"); + + /** + * Retry: every attempt, with the wait before it. + * + *

This is the {@code @Retry} annotation's retry (network-level failures and + * ProviderTransientException, 3 attempts 2s apart). It is not the + * reactive backoff in the clients that handles ProviderBusyException — that one + * logs itself, from {@code doBeforeRetry}. + */ + @Bean + public RegistryEventConsumer retryEventLogger() { + return new Consumer<>() { + @Override + public void onEntryAddedEvent(EntryAddedEvent event) { + Retry retry = event.getAddedEntry(); + log.info("retry '{}' registered — maxAttempts={}", retry.getName(), + retry.getRetryConfig().getMaxAttempts()); + + retry.getEventPublisher() + // The interesting one: an attempt failed and another is coming + // after getWaitInterval(). This is where a slow call goes. + .onRetry(e -> log.warn( + "retry '{}' attempt {} in {}ms — previous failure: {}", + e.getName(), e.getNumberOfRetryAttempts(), + e.getWaitInterval().toMillis(), describe(e.getLastThrowable()))) + // Retries are exhausted; the caller is about to see this. + .onError(e -> log.error( + "retry '{}' gave up after {} attempt(s): {}", + e.getName(), e.getNumberOfRetryAttempts(), + describe(e.getLastThrowable()))) + // Succeeded, but not on the first try — worth knowing. + .onSuccess(e -> log.info( + "retry '{}' succeeded after {} attempt(s)", + e.getName(), e.getNumberOfRetryAttempts())) + .onIgnoredError(e -> log.debug( + "retry '{}' ignoring {} (not a retryable exception)", + e.getName(), describe(e.getLastThrowable()))); + } + }; + } + + /** + * Circuit breaker: state transitions, refused calls, and the failure rate that + * caused them. + */ + @Bean + public RegistryEventConsumer circuitBreakerEventLogger() { + return new Consumer<>() { + @Override + public void onEntryAddedEvent(EntryAddedEvent event) { + CircuitBreaker breaker = event.getAddedEntry(); + log.info("circuit breaker '{}' registered — failureRateThreshold={}%, " + + "slidingWindow={}, minimumCalls={}, halfOpenCalls={}", + breaker.getName(), + breaker.getCircuitBreakerConfig().getFailureRateThreshold(), + breaker.getCircuitBreakerConfig().getSlidingWindowSize(), + breaker.getCircuitBreakerConfig().getMinimumNumberOfCalls(), + breaker.getCircuitBreakerConfig().getPermittedNumberOfCallsInHalfOpenState()); + + breaker.getEventPublisher() + .onStateTransition(e -> { + CircuitBreaker.State to = e.getStateTransition().getToState(); + CircuitBreaker.Metrics m = breaker.getMetrics(); + String detail = "%s -> %s (failureRate=%s, calls=%d, failed=%d)" + .formatted(e.getStateTransition().getFromState(), to, + failureRate(m), m.getNumberOfBufferedCalls(), + m.getNumberOfFailedCalls()); + // OPEN means every subsequent call is rejected without + // being attempted — that is an outage, not a warning. + if (to == CircuitBreaker.State.OPEN) { + log.error("circuit breaker '{}' OPEN — {}", e.getCircuitBreakerName(), detail); + } else if (to == CircuitBreaker.State.CLOSED) { + log.info("circuit breaker '{}' recovered — {}", e.getCircuitBreakerName(), detail); + } else { + log.warn("circuit breaker '{}' {}", e.getCircuitBreakerName(), detail); + } + }) + // Fired per rejected call while open: the call never reaches the + // provider at all. Deliberately does not report + // getNumberOfNotPermittedCalls() — the event fires before that + // counter is incremented, so it reads 0 while actively rejecting. + .onCallNotPermitted(e -> log.warn( + "circuit breaker '{}' rejected a call without attempting it — " + + "still OPEN, waiting out the open-state duration", + e.getCircuitBreakerName())) + .onFailureRateExceeded(e -> log.warn( + "circuit breaker '{}' failure rate {}% exceeded the threshold", + e.getCircuitBreakerName(), e.getFailureRate())) + .onSlowCallRateExceeded(e -> log.warn( + "circuit breaker '{}' slow-call rate {}% exceeded the threshold", + e.getCircuitBreakerName(), e.getSlowCallRate())) + .onError(e -> log.debug( + "circuit breaker '{}' recorded a failure after {}ms: {}", + e.getCircuitBreakerName(), e.getElapsedDuration().toMillis(), + describe(e.getThrowable()))) + // ProviderPermanentException and ProviderProcessingException are + // configured as ignored — they must not count toward the rate. + .onIgnoredError(e -> log.debug( + "circuit breaker '{}' ignoring {} (does not count toward the failure rate)", + e.getCircuitBreakerName(), describe(e.getThrowable()))) + .onSuccess(e -> log.debug("circuit breaker '{}' call ok in {}ms", + e.getCircuitBreakerName(), e.getElapsedDuration().toMillis())) + .onReset(e -> log.info("circuit breaker '{}' reset", + e.getCircuitBreakerName())); + } + }; + } + + /** + * Rate limiter: permits granted, permits refused, and how much headroom is left + * in the current refresh window. + */ + @Bean + public RegistryEventConsumer rateLimiterEventLogger() { + return new Consumer<>() { + @Override + public void onEntryAddedEvent(EntryAddedEvent event) { + RateLimiter limiter = event.getAddedEntry(); + log.info("rate limiter '{}' registered — {} call(s) per {}, waits up to {} for a permit", + limiter.getName(), + limiter.getRateLimiterConfig().getLimitForPeriod(), + limiter.getRateLimiterConfig().getLimitRefreshPeriod(), + limiter.getRateLimiterConfig().getTimeoutDuration()); + + limiter.getEventPublisher() + // A refusal means the caller already waited out the full + // timeout-duration and still got nothing — the call fails + // with RequestNotPermitted without ever reaching the provider. + .onFailure(e -> log.warn( + "rate limiter '{}' REFUSED a permit after waiting {} — " + + "call rejected before reaching the provider ({})", + e.getRateLimiterName(), + limiter.getRateLimiterConfig().getTimeoutDuration(), + permits(limiter))) + // "reserved", not "granted": on a reactive call this fires when + // the permit is RESERVED, not when the call runs. A burst of 5 + // logs five lines on the same millisecond while the calls + // themselves are still paced out one per refresh period — so + // "granted" read as though all five had just been let through. + // Per-call and high volume, hence DEBUG. + .onSuccess(e -> log.debug( + "rate limiter '{}' reserved {} permit(s) — {}", + e.getRateLimiterName(), e.getNumberOfPermits(), permits(limiter))); + } + }; + } + + /** + * Resilience4j reports -1 as "fewer calls than minimumNumberOfCalls, so the rate + * is not computed yet" — printing that raw as "-1.0%" reads like a bug. + */ + private static String failureRate(CircuitBreaker.Metrics metrics) { + float rate = metrics.getFailureRate(); + return rate < 0 ? "not yet computed" : "%.1f%%".formatted(rate); + } + + /** + * Permit headroom, in the terms the limiter actually uses. + * + *

{@code getAvailablePermissions()} goes negative under load: the + * AtomicRateLimiter hands out permits reserved against upcoming refresh windows + * rather than parking the caller, so -9 means nine calls are already spoken for + * before the next window opens. {@code getNumberOfWaitingThreads()} stays 0 for + * the same reason, which is why it is not reported — it would read as "nothing is + * queued" at the exact moment the limiter is saturated. + */ + private static String permits(RateLimiter limiter) { + int available = limiter.getMetrics().getAvailablePermissions(); + return available >= 0 + ? "%d left this window".formatted(available) + : "window exhausted, %d call(s) reserved against upcoming windows".formatted(-available); + } + + /** Exception summary that keeps the type — the message alone often says nothing. */ + private static String describe(Throwable t) { + if (t == null) { + return "none"; + } + return t.getMessage() == null + ? t.getClass().getSimpleName() + : "%s: %s".formatted(t.getClass().getSimpleName(), t.getMessage()); + } + + /** + * Only entry-added matters here — instances are never removed or replaced at + * runtime in this app. Defaulting the other two keeps each consumer to the one + * method that does something. + */ + private abstract static class Consumer implements RegistryEventConsumer { + @Override + public void onEntryRemovedEvent(EntryRemovedEvent event) { + log.info("resilience instance removed: {}", event.getRemovedEntry()); + } + + @Override + public void onEntryReplacedEvent(EntryReplacedEvent event) { + log.info("resilience instance replaced: {}", event.getNewEntry()); + } + } +} diff --git a/src/main/java/com/test/payment/controller/PaymentsController.java b/src/main/java/com/test/payment/controller/PaymentsController.java index 60a7575..b63ab5d 100644 --- a/src/main/java/com/test/payment/controller/PaymentsController.java +++ b/src/main/java/com/test/payment/controller/PaymentsController.java @@ -1,5 +1,6 @@ package com.test.payment.controller; +import com.test.payment.dto.ClientConfigDto; import com.test.payment.dto.ProviderLimitDto; import com.test.payment.models.PaymentProviderType; import com.test.payment.models.ProviderLimit; @@ -8,6 +9,8 @@ import com.test.payment.service.PaymentLifecycleService; import com.test.payment.service.PaymentLimitService; import jakarta.validation.Valid; import lombok.RequiredArgsConstructor; +import org.springframework.boot.convert.DurationStyle; +import org.springframework.core.env.Environment; import org.springframework.web.bind.annotation.GetMapping; import org.springframework.web.bind.annotation.PutMapping; import org.springframework.web.bind.annotation.RequestBody; @@ -19,6 +22,7 @@ import org.springframework.web.server.ResponseStatusException; import reactor.core.publisher.Flux; import reactor.core.publisher.Mono; +import java.time.Duration; import java.util.Arrays; @RestController @@ -28,6 +32,7 @@ public class PaymentsController { private final PaymentLifecycleService lifecycle; private final PaymentLimitService limits; + private final Environment environment; /** * All consolidated transactions across providers; optional ?provider=MPESA_KE|AIRTEL_KE|MTN_UG filter. @@ -53,6 +58,37 @@ public class PaymentsController { return limits.upsert(request); } + /** + * Settings a client should honour, read straight from {@code payments.*}. + * + *

Exists so a client does not hardcode a poll rate: the poll interval has to + * fit inside the provider's call budget, and only the server knows what that is. + */ + @GetMapping("/config") + public Mono config() { + return Mono.just(ClientConfigDto.builder() + .statusPollIntervalSeconds(seconds("payments.status-poll-interval", "15s")) + .transactionRefreshIntervalSeconds(seconds("payments.transaction-refresh-interval", "10s")) + .providerCallsPerMinute(providerCallsPerMinute()) + .build()); + } + + private long seconds(String key, String fallback) { + return DurationStyle.detectAndParse(environment.getProperty(key, fallback)).toSeconds(); + } + + /** + * Derived from the rate limiter rather than restated, so the number a client sees + * cannot drift away from the one actually being enforced. + */ + private int providerCallsPerMinute() { + long permits = environment.getProperty( + "resilience4j.ratelimiter.instances.mpesaLimiter.limit-for-period", Long.class, 1L); + Duration window = DurationStyle.detectAndParse(environment.getProperty( + "resilience4j.ratelimiter.instances.mpesaLimiter.limit-refresh-period", "2s")); + return window.isZero() ? 0 : (int) (permits * 60 / window.toSeconds()); + } + /** Null (meaning "every provider") stays null; anything else must name a real provider. */ private PaymentProviderType parseProvider(String provider) { if (provider == null) { diff --git a/src/main/java/com/test/payment/dto/ClientConfigDto.java b/src/main/java/com/test/payment/dto/ClientConfigDto.java new file mode 100644 index 0000000..4d9c75d --- /dev/null +++ b/src/main/java/com/test/payment/dto/ClientConfigDto.java @@ -0,0 +1,34 @@ +package com.test.payment.dto; + +import lombok.AllArgsConstructor; +import lombok.Builder; +import lombok.Data; +import lombok.NoArgsConstructor; + +/** + * Client-facing settings, served by {@code GET /api/payments/config}. + * + *

These live on the server because the server is what knows the provider's rate + * limits. A client that picks its own poll rate is guessing at a budget it cannot + * see: polling a PENDING payment is a live provider call, so a client polling too + * eagerly spends the same quota the payments themselves need, and the first symptom + * is a 429 on someone else's payment. + * + *

Advisory, not enforced — nothing stops a client ignoring these. The rate limiter + * is what actually protects the provider. + */ +@Data +@Builder +@AllArgsConstructor +@NoArgsConstructor +public class ClientConfigDto { + + /** How often to re-check a payment that is still PENDING. Costs a provider call. */ + private long statusPollIntervalSeconds; + + /** How often to refresh the transaction list. Database only, no provider call. */ + private long transactionRefreshIntervalSeconds; + + /** Calls per minute the provider allows, so a client can show its budget. */ + private int providerCallsPerMinute; +} diff --git a/src/main/java/com/test/payment/jobs/PaymentReconciliationJob.java b/src/main/java/com/test/payment/jobs/PaymentReconciliationJob.java index c8548d9..9f94dd9 100644 --- a/src/main/java/com/test/payment/jobs/PaymentReconciliationJob.java +++ b/src/main/java/com/test/payment/jobs/PaymentReconciliationJob.java @@ -50,6 +50,8 @@ public class PaymentReconciliationJob { LocalDateTime now = LocalDateTime.now(); Duration pendingAge = duration("payments.reconciliation.pending-age", "3m"); Duration staleAge = duration("payments.reconciliation.stale-age", "3h"); + int maxPerRun = environment.getProperty("payments.reconciliation.max-per-run", Integer.class, 10); + LocalDateTime pendingCutoff = now.minus(pendingAge); // Order matters: close off the hopeless ones first so the re-query pass below // does not keep hammering the operator for payments we have given up on. @@ -59,7 +61,21 @@ public class PaymentReconciliationJob { log.warn("{} initiation(s) still pending after {} — marked Unresolved", count, staleAge); } }) - .thenMany(lifecycle.findPendingOlderThan(now.minus(pendingAge))) + // Capped, oldest first. Uncapped, a backlog of N open payments queues N + // provider status queries in one tick; at the operator's pace that is + // minutes of work per 60-second tick, and it holds the entire call + // budget indefinitely while real payments queue behind it. + .thenMany(lifecycle.countPendingOlderThan(pendingCutoff) + .doOnNext(waiting -> { + if (waiting > maxPerRun) { + // Say what was left behind — a silent cap reads as + // "everything was reconciled" when it was not. + log.info("{} initiation(s) pending past {} — reconciling the {} oldest this run, " + + "{} deferred to later runs", + waiting, pendingAge, maxPerRun, waiting - maxPerRun); + } + }) + .thenMany(lifecycle.findPendingOlderThan(pendingCutoff, maxPerRun))) .concatMap(initiation -> { PaymentProviderService service = servicesByProvider.get(initiation.getProvider()); if (service == null) { diff --git a/src/main/java/com/test/payment/repository/PaymentInitiationRepository.java b/src/main/java/com/test/payment/repository/PaymentInitiationRepository.java index be4962e..d839ea0 100644 --- a/src/main/java/com/test/payment/repository/PaymentInitiationRepository.java +++ b/src/main/java/com/test/payment/repository/PaymentInitiationRepository.java @@ -2,6 +2,7 @@ package com.test.payment.repository; import com.test.payment.models.PaymentInitiation; import com.test.payment.models.PaymentProviderType; +import org.springframework.data.domain.Pageable; import org.springframework.data.jpa.repository.JpaRepository; import org.springframework.data.jpa.repository.Query; import org.springframework.data.repository.query.Param; @@ -18,12 +19,31 @@ import java.util.List; @Repository public interface PaymentInitiationRepository extends JpaRepository { + /** + * Oldest first, and bounded by the caller's {@link Pageable}. + * + *

Both matter for reconciliation. Unbounded, one tick queues a provider status + * query for every open payment — with a backlog of 60 that is minutes of work per + * 60-second tick, and it consumes the operator's whole call budget indefinitely. + * Oldest-first makes the backlog drain FIFO instead of re-querying the same head + * of the list every tick. Pass {@code Pageable.unpaged()} for the database-only + * sweeps, where there is no per-row cost to bound. + */ @Query(""" SELECT i FROM PaymentInitiation i WHERE i.Status.Name = :statusName AND i.CreatedAt < :cutoff + ORDER BY i.CreatedAt ASC """) List findByStatusNameAndCreatedAtBefore(@Param("statusName") String statusName, - @Param("cutoff") LocalDateTime cutoff); + @Param("cutoff") LocalDateTime cutoff, + Pageable page); + + @Query(""" + SELECT count(i) FROM PaymentInitiation i + WHERE i.Status.Name = :statusName AND i.CreatedAt < :cutoff + """) + long countByStatusNameAndCreatedAtBefore(@Param("statusName") String statusName, + @Param("cutoff") LocalDateTime cutoff); /** * What this payer has already committed to on this provider inside the window. diff --git a/src/main/java/com/test/payment/service/AirtelService.java b/src/main/java/com/test/payment/service/AirtelService.java index 5df1e2c..63a91f7 100644 --- a/src/main/java/com/test/payment/service/AirtelService.java +++ b/src/main/java/com/test/payment/service/AirtelService.java @@ -65,7 +65,7 @@ public class AirtelService implements PaymentProviderService { .map(response -> toResponseData(response, reference)); }) .flatMap(data -> lifecycle.persistResponse(initiation, data)) - .onErrorResume(ex -> lifecycle.markFailed(initiation, ex))); + .onErrorResume(ex -> lifecycle.markUnsuccessful(initiation, ex))); } public Mono handleCallback(AirtelCallbackPayload payload) { diff --git a/src/main/java/com/test/payment/service/MpesaService.java b/src/main/java/com/test/payment/service/MpesaService.java index 5a195b4..a6bb2be 100644 --- a/src/main/java/com/test/payment/service/MpesaService.java +++ b/src/main/java/com/test/payment/service/MpesaService.java @@ -65,7 +65,7 @@ public class MpesaService implements PaymentProviderService { .doOnError(ex -> audit.complete(call, null, null, ex)); }) .flatMap(response -> lifecycle.persistResponse(initiation, toResponseData(response))) - .onErrorResume(ex -> lifecycle.markFailed(initiation, ex))); + .onErrorResume(ex -> lifecycle.markUnsuccessful(initiation, ex))); } public Mono handleCallback(StkCallbackPayload payload) { diff --git a/src/main/java/com/test/payment/service/MtnService.java b/src/main/java/com/test/payment/service/MtnService.java index 117813b..01c7f08 100644 --- a/src/main/java/com/test/payment/service/MtnService.java +++ b/src/main/java/com/test/payment/service/MtnService.java @@ -67,7 +67,7 @@ public class MtnService implements PaymentProviderService { reference, null, "202", "Accepted", "Request to pay accepted", true)); }) .flatMap(data -> lifecycle.persistResponse(initiation, data)) - .onErrorResume(ex -> lifecycle.markFailed(initiation, ex))); + .onErrorResume(ex -> lifecycle.markUnsuccessful(initiation, ex))); } public Mono handleCallback(MtnStatusResponseDto payload) { diff --git a/src/main/java/com/test/payment/service/PaymentLifecycleService.java b/src/main/java/com/test/payment/service/PaymentLifecycleService.java index 0ab90cf..e283b4b 100644 --- a/src/main/java/com/test/payment/service/PaymentLifecycleService.java +++ b/src/main/java/com/test/payment/service/PaymentLifecycleService.java @@ -8,6 +8,8 @@ import com.test.payment.models.PaymentInitiation; import com.test.payment.models.PaymentProviderType; import com.test.payment.models.Status; import com.test.payment.models.Transaction; +import io.github.resilience4j.circuitbreaker.CallNotPermittedException; +import io.github.resilience4j.ratelimiter.RequestNotPermitted; import lombok.RequiredArgsConstructor; import lombok.extern.slf4j.Slf4j; import org.springframework.dao.OptimisticLockingFailureException; @@ -75,15 +77,42 @@ public class PaymentLifecycleService { return blocking(() -> store.persistResponse(initiation.getId(), data)); } - public Mono markFailed(PaymentInitiation initiation, Throwable ex) { - log.error("[{}] payment failed for initiation {}: {}", initiation.getProvider(), initiation.getId(), ex.toString()); + /** + * Records a payment that did not succeed, distinguishing two cases that look the + * same to the caller but are not the same thing at all. + * + *

If one of our own guards refused the call — the circuit breaker is open, or the + * rate limiter would not issue a permit — the operator never saw the request and no + * payment was attempted. That is Rejected, not Failed. Recording it as Failed + * produces rows that read as failed payments when nothing was sent, and they carry + * no provider reference, so nothing can ever resolve them. + * + *

Anything else did reach the operator, so it is a real failure. + */ + public Mono markUnsuccessful(PaymentInitiation initiation, Throwable ex) { String reason = ex.getMessage() == null ? ex.getClass().getSimpleName() : ex.getMessage(); + + if (neverDispatched(ex)) { + log.warn("[{}] initiation {} never sent — {}", + initiation.getProvider(), initiation.getId(), ex.getClass().getSimpleName()); + return blocking(() -> { + store.markNotDispatched(initiation.getId(), reason); + return true; + }).then(Mono.error(ex)); + } + + log.error("[{}] payment failed for initiation {}: {}", initiation.getProvider(), initiation.getId(), ex.toString()); return blocking(() -> { store.markFailed(initiation.getId(), reason); return true; }).then(Mono.error(ex)); } + /** Resilience4j threw before the call was made, so nothing reached the provider. */ + private static boolean neverDispatched(Throwable ex) { + return ex instanceof CallNotPermittedException || ex instanceof RequestNotPermitted; + } + public Mono applyCallback(PaymentProviderType provider, CallbackData data, String rawPayload) { if (data.providerReference() == null) { log.warn("[{}] callback without a provider reference ignored", provider); @@ -131,9 +160,18 @@ public class PaymentLifecycleService { return blocking(() -> store.markStaleUnresolved(cutoff)); } - /** PENDING initiations older than the cutoff, for the reconciliation job. */ - public Flux findPendingOlderThan(LocalDateTime cutoff) { - return blocking(() -> store.findPendingOlderThan(cutoff)).flatMapMany(Flux::fromIterable); + /** + * PENDING initiations older than the cutoff, oldest first, at most {@code limit}. + * Every row costs one live provider call, so the cap is the reconciliation job's + * grip on the operator's budget. + */ + public Flux findPendingOlderThan(LocalDateTime cutoff, int limit) { + return blocking(() -> store.findPendingOlderThan(cutoff, limit)).flatMapMany(Flux::fromIterable); + } + + /** Total waiting, so a capped run can report what it did not get to. */ + public Mono countPendingOlderThan(LocalDateTime cutoff) { + return blocking(() -> store.countPendingOlderThan(cutoff)); } /** diff --git a/src/main/java/com/test/payment/service/PaymentLifecycleStore.java b/src/main/java/com/test/payment/service/PaymentLifecycleStore.java index 8d59157..3c23a02 100644 --- a/src/main/java/com/test/payment/service/PaymentLifecycleStore.java +++ b/src/main/java/com/test/payment/service/PaymentLifecycleStore.java @@ -33,6 +33,8 @@ import lombok.RequiredArgsConstructor; import lombok.extern.slf4j.Slf4j; import org.springframework.dao.DataIntegrityViolationException; import org.springframework.http.HttpStatus; +import org.springframework.data.domain.PageRequest; +import org.springframework.data.domain.Pageable; import org.springframework.stereotype.Service; import org.springframework.transaction.annotation.Transactional; import org.springframework.web.server.ResponseStatusException; @@ -154,6 +156,21 @@ public class PaymentLifecycleStore { recordTransaction(updated, null, null, truncate(reason), null, null, "ERROR"); } + /** + * The request never left this application — an open circuit breaker or a refused + * rate-limiter permit turned it away. + * + *

Separate from {@link #markFailed} on purpose. NOT_SENT is distinguishable from + * ERROR (the operator was called and something went wrong) and from REJECTION (the + * operator was called and declined the request), so a burst of locally-rejected + * calls cannot be mistaken for a wave of failing payments. + */ + @Transactional + public void markNotDispatched(Long initiationId, String reason) { + PaymentInitiation updated = updateStatus(requireInitiation(initiationId), statuses.rejected()); + recordTransaction(updated, null, null, truncate(reason), null, null, "NOT_SENT"); + } + @Transactional public CallbackAckDto applyCallback(PaymentProviderType provider, CallbackData data, String rawPayload) { Optional match = findResponseByReference(provider, data.providerReference()); @@ -251,8 +268,10 @@ public class PaymentLifecycleStore { */ @Transactional public int markStaleUnresolved(LocalDateTime cutoff) { - List stale = - initiationRepository.findByStatusNameAndCreatedAtBefore(statuses.pending().getName(), cutoff); + // Unbounded on purpose: this pass only writes rows, it makes no provider call, + // so there is no budget to protect by capping it. + List stale = initiationRepository.findByStatusNameAndCreatedAtBefore( + statuses.pending().getName(), cutoff, Pageable.unpaged()); for (PaymentInitiation initiation : stale) { PaymentInitiation updated = updateStatus(initiation, statuses.unresolved()); @@ -267,9 +286,18 @@ public class PaymentLifecycleStore { return stale.size(); } + /** Oldest-first and capped: every row returned costs one live provider call. */ @Transactional(readOnly = true) - public List findPendingOlderThan(LocalDateTime cutoff) { - return initiationRepository.findByStatusNameAndCreatedAtBefore(statuses.pending().getName(), cutoff); + public List findPendingOlderThan(LocalDateTime cutoff, int limit) { + return initiationRepository.findByStatusNameAndCreatedAtBefore( + statuses.pending().getName(), cutoff, PageRequest.of(0, limit)); + } + + /** How many are waiting, so a capped sweep can say what it left behind. */ + @Transactional(readOnly = true) + public long countPendingOlderThan(LocalDateTime cutoff) { + return initiationRepository.countByStatusNameAndCreatedAtBefore( + statuses.pending().getName(), cutoff); } // --- per-operator dispatch ------------------------------------------------- diff --git a/src/main/java/com/test/payment/service/StatusCatalog.java b/src/main/java/com/test/payment/service/StatusCatalog.java index a200d7d..2cfb213 100644 --- a/src/main/java/com/test/payment/service/StatusCatalog.java +++ b/src/main/java/com/test/payment/service/StatusCatalog.java @@ -31,6 +31,7 @@ public class StatusCatalog { private static final String PENDING = "Pending"; private static final String FAILED = "Failed"; private static final String UNRESOLVED = "Unresolved"; + private static final String REJECTED = "Rejected"; /** * What {@code statuses} is seeded with when a state is missing. The table is the @@ -42,7 +43,9 @@ public class StatusCatalog { new StatusDefinition(PENDING, "Pushed to the payer, awaiting confirmation or the provider callback"), new StatusDefinition(FAILED, "Rejected by the provider, declined by the payer, or timed out"), new StatusDefinition(UNRESOLVED, "Still pending past the reconciliation window — the operator never " - + "reported an outcome, so whether money moved is unknown and needs manual follow-up")); + + "reported an outcome, so whether money moved is unknown and needs manual follow-up"), + new StatusDefinition(REJECTED, "Never sent to the operator — a local guard (open circuit breaker or " + + "refused rate-limiter permit) turned it away, so no payment was ever attempted")); /** * Least to most informed. A payment may only move up this order, which is @@ -50,7 +53,8 @@ public class StatusCatalog { * status: whichever arrives second cannot undo a better-informed verdict, but a * receipt-bearing callback can still correct a pessimistic query result. */ - private static final List PRECEDENCE = List.of(PENDING, UNRESOLVED, FAILED, SUCCESS, PAID); + private static final List PRECEDENCE = + List.of(PENDING, UNRESOLVED, REJECTED, FAILED, SUCCESS, PAID); private final StatusRepository statusRepository; private final Map byName = new ConcurrentHashMap<>(); @@ -88,6 +92,19 @@ public class StatusCatalog { return require(UNRESOLVED); } + /** + * Never dispatched: the request was turned away by one of our own guards before it + * reached the operator. + * + *

Deliberately not Failed. Failed means the operator saw the payment and it did + * not succeed; this means the operator never saw it at all. Recording these as + * Failed produces rows that read as failed payments when nothing was ever sent — + * and there is no provider reference on them, so nothing can ever resolve them. + */ + public Status rejected() { + return require(REJECTED); + } + /** * Whether a payment may move to {@code target}. Guards against duplicate and * out-of-order status updates: re-applying the state it already has is a no-op, diff --git a/src/main/resources/application.yml b/src/main/resources/application.yml index fa272a9..e9de09d 100644 --- a/src/main/resources/application.yml +++ b/src/main/resources/application.yml @@ -5,9 +5,25 @@ resilience4j: ratelimiter: instances: mpesaLimiter: - limit-for-period: 5 - limit-refresh-period: 1s - timeout-duration: 5s + # Calibrated to Safaricom's actual Apigee spike arrest, observed live on the + # STK query endpoint: "messagesPerPeriod=30, periodInMicroseconds=60000000, + # maxBurstMessageCount=3" — 30 calls a minute, i.e. one every 2 seconds. + # + # Spike arrest enforces SPACING, not a bucket you drain, so this is 1-per-2s + # rather than 30-per-60s: the latter would let 30 calls fire at the top of the + # minute and every one after the third would come back 429. + # + # The previous 5/s was 300/min — 10x more permissive than the provider — so + # the limiter granted every permit right up until Safaricom rejected the call. + # A local limiter looser than the remote one protects nothing. + # + # Sandbox and production quotas differ, and this was measured on the query + # endpoint; raise it if your production API product is provisioned higher. + limit-for-period: 1 + limit-refresh-period: 2s + # A permit now takes up to 2s to arrive, so the wait has to exceed that or + # calls would be refused while a permit was about to free up. + timeout-duration: 10s airtelLimiter: limit-for-period: 5 limit-refresh-period: 1s @@ -43,6 +59,11 @@ resilience4j: ignore-exceptions: - com.test.payment.exceptions.ProviderPermanentException - com.test.payment.exceptions.ProviderProcessingException + # Being throttled is not the provider being unhealthy — it answered, promptly + # and correctly, to say "slow down". Counting 429s as failures turns a rate + # limit into an outage: the breaker opens, and calls we would otherwise have + # paced correctly are rejected without being attempted. + - com.test.payment.exceptions.ProviderBusyException airtelCircuitBreaker: *provider-circuit-breaker mtnCircuitBreaker: *provider-circuit-breaker @@ -68,7 +89,19 @@ spring: logging: level: - org.hibernate.SQL: DEBUG + # Every statement Hibernate issues — off by default because it is one line per + # query and drowns everything else. Uncomment to debug a query. + # org.hibernate.SQL: DEBUG + # Retry attempts, circuit-breaker state changes, rate-limiter permits and the + # ProviderBusyException backoff all log through this one name + # (ResilienceEventLogging + ProviderBackoff). INFO gives registrations, retries, + # refusals and state transitions; DEBUG adds the per-call chatter — every permit + # granted with the remaining headroom in the refresh window, and every recorded + # success/failure the breaker counts. + com.test.payment.resilience: DEBUG + # Token tier hits: served from Redis / from the database / fetched fresh. + # The fetch itself is INFO; which tier answered is DEBUG. + com.test.payment.service.TokenCacheService: DEBUG springdoc: swagger-ui: @@ -81,11 +114,27 @@ springdoc: payments: token-expiry-buffer-seconds: 60 + # How often a client should re-check a payment that is still PENDING. + # Served to clients by GET /api/payments/config so the poll rate is set here, + # next to the rate limits it has to live within, rather than hardcoded per client. + # + # Every poll of a PENDING payment is a live provider call, so this spends the same + # 30/min budget as the payments themselves: at 15s one open payment costs 4 calls a + # minute. The 5s it replaces cost 12 — 40% of the budget on a single idle payment. + status-poll-interval: 15s + # Consolidated-transaction refresh. A database read, no provider call, so it is + # only bounded by how much load you want on your own app. + transaction-refresh-interval: 10s reconciliation: # payments still Pending after this age are re-queried against the provider pending-age: 3m # after this long with no outcome we stop querying and mark them Unresolved stale-age: 3h + # Most payments one run will re-query. Each one is a live provider call, so this is + # the job's share of the operator's budget: at 10 per 60s tick it takes a third of + # the 30/min ceiling and leaves the rest for actual payments. Uncapped, a backlog + # of 60 open payments queued 60 calls per tick and starved everything else. + max-per-run: 10 # how often the reconciliation job runs fixed-delay: 60s