Guard status transitions and close off stale pending payments

- StatusCatalog gains an Unresolved state and a PRECEDENCE order
  (Pending -> Unresolved -> Failed -> Success -> Paid); canTransition only
  allows moves up it, so a callback and a status query racing each other
  cannot make the status flap. A refused transition still stores the row.
- PaymentInitiation carries a @Version optimistic lock; callback and query
  paths go through blockingWithRetry, which replays the losing unit of work
  once instead of dropping it.
- Reconciliation runs two passes per tick, stale first: anything Pending past
  payments.reconciliation.stale-age (3h) is marked Unresolved with
  resolvedBy = EXPIRY, then anything past pending-age (now 3m) is re-queried.
- Explicit lowercase @Table names on the entities that were relying on
  derived naming.

Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>

CLAUDE.md

@@ -79,7 +79,11 @@ Spring Boot 4 / Java 21 reactive (WebFlux) multi-provider mobile-money service (
 - `PROVIDER_LIMITS` — configurable payment ceilings, `UNIQUE (provider, period, scope)`. Periods come from the `LimitPeriod` enum (PER_TRANSACTION, DAILY, MONTHLY — adding a constant is all a new period needs); `scope` from the `LimitScope` enum (PER_PAYER buckets by paying MSISDN, MERCHANT sums every payer on the provider; defaults to PER_PAYER and is ignored by PER_TRANSACTION). Defaults are seeded by `DataSeeder` only when the triple is absent (and only for the markets `<operator>.country` actually configures), so runtime edits survive a restart. `PaymentLimitService.enforce` runs in every `<Provider>Service.initiatePayment` **before** the initiation is persisted (breaches leave no DB row) and raises `PaymentLimitExceededException` → 422 `LIMIT_EXCEEDED`. Cumulative periods sum non-Failed initiations inside the window (per MSISDN for PER_PAYER, provider-wide for MERCHANT), so PENDING pushes count against the cap. An unrecognised `scope` on a row falls back to PER_PAYER (the tighter interpretation).
 - **Provider-call audit trail** (`models/audit`, `repository/audit`, `ProviderCallAudit`): every outbound call and inbound callback is recorded per operator in its own independent tables — `mpesa_requests` / `mpesa_responses` / `mpesa_callback_responses`, and the `airtel_*` and `mtn_*` equivalents. Responses and callbacks reference their request `@ManyToOne`. The nine entities are **fully standalone** — no shared supertype, no `@MappedSuperclass`, no discriminator, no join. `ProviderCallAudit` therefore dispatches on `Operator` with an explicit branch per provider rather than polymorphically, which is verbose on purpose: a new operator will not compile until all three of its tables are wired up. Writes run on the `auditExecutor` pool (bounded queue, drop-on-overflow, daemon threads) so a payment is never delayed or failed by its audit trail; ordering within a call is kept by chaining onto the request's `CompletableFuture` rather than blocking on it. Credentials matching `ProviderCallAudit.SECRETS` (Password, passkey, api-key, secret, authorization, access_token) are masked before anything is stored.
 - `@Table` names must be lowercase — Postgres folds unquoted DDL identifiers to lowercase and Spring Data quotes an explicit entity name verbatim. Column names carry no `@Column` annotation, so they are derived and adapt to the dialect's casing on their own; keep it that way.
-- `PaymentReconciliationJob` reconciles PENDING initiations of **all** providers older than `payments.reconciliation.pending-age` (default 5m) by dispatching to the right `PaymentProviderService`; interval `payments.reconciliation.fixed-delay` (default 60s).
+- `PaymentReconciliationJob` runs two passes per tick (interval `payments.reconciliation.fixed-delay`, default 60s), **stale first** so hopeless payments stop being re-queried:
+  1. anything still Pending past `payments.reconciliation.stale-age` (default 3h) → `Unresolved` via `markStaleUnresolved`, `resolvedBy = EXPIRY`;
+  2. anything still Pending past `payments.reconciliation.pending-age` (default **3m**) → live provider status query, dispatched to the right `PaymentProviderService`.
+- **Status transitions are guarded, not blind.** `StatusCatalog.PRECEDENCE` orders the states least- to most-informed (Pending → Unresolved → Failed → Success → Paid) and `canTransition` only allows moves *up* it. So a callback and a status query racing each other cannot make the status flap: re-asserting the current state is a silent no-op (DEBUG), a worse verdict is refused with a WARN, and a receipt-bearing callback can still rescue a payment the query gave up on (Unresolved → Paid). **A refused transition never discards the row** — the callback is still stored, only the status is held back.
+- `PaymentInitiation` carries a `@Version` optimistic lock. When a callback and a query genuinely commit at the same instant the loser's transaction rolls back; `PaymentLifecycleService.blockingWithRetry` replays it once, so the callback row still lands and the guard then declines the redundant status change. Use it for any new path that resolves a payment.
 
 **Configuration (`application.yml`):**
 - Config is read via `Environment.getProperty` by project convention (no `@ConfigurationProperties`).

src/main/java/com/test/payment/jobs/PaymentReconciliationJob.java

@@ -47,11 +47,19 @@ public class PaymentReconciliationJob {
             log.debug("Previous reconciliation run still in progress — skipping this tick");
             return;
         }
-        Duration pendingAge = DurationStyle.detectAndParse(
-                environment.getProperty("payments.reconciliation.pending-age", "5m"));
-        LocalDateTime cutoff = LocalDateTime.now().minus(pendingAge);
+        LocalDateTime now = LocalDateTime.now();
+        Duration pendingAge = duration("payments.reconciliation.pending-age", "3m");
+        Duration staleAge = duration("payments.reconciliation.stale-age", "3h");
 
-        lifecycle.findPendingOlderThan(cutoff)
+        // 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.
+        lifecycle.markStaleUnresolved(now.minus(staleAge))
+                .doOnNext(count -> {
+                    if (count > 0) {
+                        log.warn("{} initiation(s) still pending after {} — marked Unresolved", count, staleAge);
+                    }
+                })
+                .thenMany(lifecycle.findPendingOlderThan(now.minus(pendingAge)))
                 .concatMap(initiation -> {
                     PaymentProviderService service = servicesByProvider.get(initiation.getProvider());
                     if (service == null) {
@@ -73,4 +81,8 @@ public class PaymentReconciliationJob {
                         v -> { },
                         e -> log.error("Reconciliation run aborted: {}", e.toString()));
     }
+
+    private Duration duration(String key, String fallback) {
+        return DurationStyle.detectAndParse(environment.getProperty(key, fallback));
+    }
 }

src/main/java/com/test/payment/models/AirtelPaymentCallback.java

@@ -27,6 +27,7 @@ import java.time.LocalDateTime;
  * duplicate-inclusive record lives in airtel_callback_responses instead.
  */
 @Entity
+@Table(name = "airtel_payment_callbacks")
 @Getter
 @Setter
 @Builder

src/main/java/com/test/payment/models/AirtelPaymentResponse.java

@@ -27,6 +27,7 @@ import java.time.LocalDateTime;
  * the lifecycle reads it through PaymentLifecycleStore's StoredResponse view.
  */
 @Entity
+@Table(name = "airtel_payment_responses")
 @Getter
 @Setter
 @Builder

src/main/java/com/test/payment/models/MpesaPaymentCallback.java

@@ -27,6 +27,7 @@ import java.time.LocalDateTime;
  * duplicate-inclusive record lives in mpesa_callback_responses instead.
  */
 @Entity
+@Table(name = "mpesa_payment_callbacks")
 @Getter
 @Setter
 @Builder

src/main/java/com/test/payment/models/MpesaPaymentResponse.java

@@ -27,6 +27,7 @@ import java.time.LocalDateTime;
  * the lifecycle reads it through PaymentLifecycleStore's StoredResponse view.
  */
 @Entity
+@Table(name = "mpesa_payment_responses")
 @Getter
 @Setter
 @Builder

src/main/java/com/test/payment/models/MtnPaymentCallback.java

@@ -27,6 +27,7 @@ import java.time.LocalDateTime;
  * duplicate-inclusive record lives in mtn_callback_responses instead.
  */
 @Entity
+@Table(name = "mtn_payment_callbacks")
 @Getter
 @Setter
 @Builder

src/main/java/com/test/payment/models/MtnPaymentResponse.java

@@ -27,6 +27,7 @@ import java.time.LocalDateTime;
  * the lifecycle reads it through PaymentLifecycleStore's StoredResponse view.
  */
 @Entity
+@Table(name = "mtn_payment_responses")
 @Getter
 @Setter
 @Builder

src/main/java/com/test/payment/models/PaymentInitiation.java

@@ -12,6 +12,7 @@ import jakarta.persistence.Index;
 import jakarta.persistence.JoinColumn;
 import jakarta.persistence.ManyToOne;
 import jakarta.persistence.Table;
+import jakarta.persistence.Version;
 import lombok.AllArgsConstructor;
 import lombok.Builder;
 import lombok.Getter;
@@ -23,6 +24,8 @@ import java.math.BigDecimal;
 import java.time.LocalDateTime;
 
 @Entity
+@Table(name = "payment_initiations",
+        indexes = @Index(name = "idx_initiations_status_created", columnList = "Status, CreatedAt"))
 @Getter
 @Setter
 @Builder
@@ -58,6 +61,14 @@ public class PaymentInitiation {
     @JoinColumn(name = "Status", nullable = false)
     private Status Status;
 
+    /**
+     * Optimistic lock. A callback and a status query can resolve the same payment at
+     * the same instant; without this the second write would silently clobber the
+     * first. With it, the loser fails fast and is handled as a concurrent resolution.
+     */
+    @Version
+    private Long Version;
+
     @Column(nullable = false)
     private LocalDateTime CreatedAt;
 

src/main/java/com/test/payment/models/ProviderToken.java

@@ -22,6 +22,7 @@ import org.hibernate.type.SqlTypes;
 import java.time.LocalDateTime;
 
 @Entity
+@Table(name = "provider_tokens")
 @Getter
 @Setter
 @Builder

src/main/java/com/test/payment/models/Transaction.java

@@ -26,6 +26,7 @@ import java.time.LocalDateTime;
 
 
 @Entity
+@Table(name = "transactions")
 @Getter
 @Setter
 @Builder

src/main/java/com/test/payment/models/audit/AirtelCallbackResponse.java

@@ -26,6 +26,7 @@ import java.time.Instant;
  * that silently drops repeats is not one.
  */
 @Entity
+@Table(name = "airtel_callback_responses")
 @Getter
 @Setter
 @ToString(exclude = "Request")

src/main/java/com/test/payment/models/audit/AirtelRequest.java

@@ -21,6 +21,7 @@ import java.time.Instant;
  * waits for this row.
  */
 @Entity
+@Table(name = "airtel_requests")
 @Getter
 @Setter
 @ToString

src/main/java/com/test/payment/models/audit/AirtelResponse.java

@@ -20,6 +20,7 @@ import java.time.Instant;
  * usually what an audit is actually needed for.
  */
 @Entity
+@Table(name = "airtel_responses")
 @Getter
 @Setter
 @ToString(exclude = "Request")

src/main/java/com/test/payment/models/audit/MpesaCallbackResponse.java

@@ -26,6 +26,7 @@ import java.time.Instant;
  * that silently drops repeats is not one.
  */
 @Entity
+@Table(name = "mpesa_callback_responses")
 @Getter
 @Setter
 @ToString(exclude = "Request")

src/main/java/com/test/payment/models/audit/MpesaRequest.java

@@ -21,6 +21,7 @@ import java.time.Instant;
  * waits for this row.
  */
 @Entity
+@Table(name = "mpesa_requests")
 @Getter
 @Setter
 @ToString

src/main/java/com/test/payment/models/audit/MpesaResponse.java

@@ -20,6 +20,7 @@ import java.time.Instant;
  * usually what an audit is actually needed for.
  */
 @Entity
+@Table(name = "mpesa_responses")
 @Getter
 @Setter
 @ToString(exclude = "Request")

src/main/java/com/test/payment/models/audit/MtnCallbackResponse.java

@@ -26,6 +26,7 @@ import java.time.Instant;
  * that silently drops repeats is not one.
  */
 @Entity
+@Table(name = "mtn_callback_responses")
 @Getter
 @Setter
 @ToString(exclude = "Request")

src/main/java/com/test/payment/models/audit/MtnRequest.java

@@ -21,6 +21,7 @@ import java.time.Instant;
  * waits for this row.
  */
 @Entity
+@Table(name = "mtn_requests")
 @Getter
 @Setter
 @ToString

src/main/java/com/test/payment/models/audit/MtnResponse.java

@@ -20,6 +20,7 @@ import java.time.Instant;
  * usually what an audit is actually needed for.
  */
 @Entity
+@Table(name = "mtn_responses")
 @Getter
 @Setter
 @ToString(exclude = "Request")

src/main/java/com/test/payment/service/PaymentLifecycleService.java

@@ -10,10 +10,12 @@ import com.test.payment.models.Status;
 import com.test.payment.models.Transaction;
 import lombok.RequiredArgsConstructor;
 import lombok.extern.slf4j.Slf4j;
+import org.springframework.dao.OptimisticLockingFailureException;
 import org.springframework.stereotype.Service;
 import reactor.core.publisher.Flux;
 import reactor.core.publisher.Mono;
 import reactor.core.scheduler.Schedulers;
+import reactor.util.retry.Retry;
 
 import java.math.BigDecimal;
 import java.time.LocalDateTime;
@@ -87,7 +89,7 @@ public class PaymentLifecycleService {
             log.warn("[{}] callback without a provider reference ignored", provider);
             return Mono.just(CallbackAckDto.accepted("Ignored: no reference"));
         }
-        return blocking(() -> store.applyCallback(provider, data, rawPayload));
+        return blockingWithRetry(() -> store.applyCallback(provider, data, rawPayload));
     }
 
     /**
@@ -100,7 +102,7 @@ public class PaymentLifecycleService {
         return blocking(() -> store.loadForStatusCheck(provider, providerReference))
                 .flatMap(context -> context.pending()
                         ? querier.apply(context.response())
-                                .flatMap(outcome -> blocking(() ->
+                                .flatMap(outcome -> blockingWithRetry(() ->
                                         store.applyQueryOutcome(context.response().initiationId(), outcome)))
                         : Mono.just(context.currentState()));
     }
@@ -121,6 +123,14 @@ public class PaymentLifecycleService {
         return blocking(() -> store.listTransactions(provider)).flatMapMany(Flux::fromIterable);
     }
 
+    /**
+     * Closes off payments still pending past the stale cutoff as Unresolved.
+     * Runs before the re-query pass, so they are not queried again.
+     */
+    public Mono<Integer> markStaleUnresolved(LocalDateTime cutoff) {
+        return blocking(() -> store.markStaleUnresolved(cutoff));
+    }
+
     /** PENDING initiations older than the cutoff, for the reconciliation job. */
     public Flux<PaymentInitiation> findPendingOlderThan(LocalDateTime cutoff) {
         return blocking(() -> store.findPendingOlderThan(cutoff)).flatMapMany(Flux::fromIterable);
@@ -133,4 +143,20 @@ public class PaymentLifecycleService {
     private <T> Mono<T> blocking(Callable<T> work) {
         return Mono.fromCallable(work).subscribeOn(Schedulers.boundedElastic());
     }
+
+    /**
+     * As {@link #blocking}, but retried once when the optimistic lock on the initiation
+     * is lost. That happens when a callback and a status query resolve the same payment
+     * at the same instant: the loser's transaction rolls back, and replaying it lets the
+     * callback row still be written while the status guard declines the now-redundant
+     * update. Without the retry the loser would simply be dropped.
+     */
+    private <T> Mono<T> blockingWithRetry(Callable<T> work) {
+        return blocking(work)
+                .retryWhen(Retry.max(1)
+                        .filter(OptimisticLockingFailureException.class::isInstance)
+                        .doBeforeRetry(signal -> log.info(
+                                "Concurrent resolution detected ({}) — replaying the unit of work",
+                                signal.failure().getClass().getSimpleName())));
+    }
 }

src/main/java/com/test/payment/service/PaymentLifecycleStore.java

@@ -244,6 +244,29 @@ public class PaymentLifecycleStore {
                 : transactionRepository.findByProvider(provider);
     }
 
+    /**
+     * Gives up on payments the operator never resolved: anything still Pending past
+     * the stale cutoff becomes Unresolved, so it stops being re-queried forever and
+     * shows up as needing manual follow-up. Returns how many were closed off.
+     */
+    @Transactional
+    public int markStaleUnresolved(LocalDateTime cutoff) {
+        List<PaymentInitiation> stale =
+                initiationRepository.findByStatusNameAndCreatedAtBefore(statuses.pending().getName(), cutoff);
+
+        for (PaymentInitiation initiation : stale) {
+            PaymentInitiation updated = updateStatus(initiation, statuses.unresolved());
+            StoredResponse response =
+                    findResponseByInitiation(initiation.getProvider(), initiation.getId()).orElse(null);
+            recordTransaction(updated, response, null,
+                    "Still pending past the reconciliation window — outcome never reported",
+                    null, null, "EXPIRY");
+            log.warn("[{}] initiation {} unresolved after {} — no outcome from the operator",
+                    initiation.getProvider(), initiation.getId(), initiation.getCreatedAt());
+        }
+        return stale.size();
+    }
+
     @Transactional(readOnly = true)
     public List<PaymentInitiation> findPendingOlderThan(LocalDateTime cutoff) {
         return initiationRepository.findByStatusNameAndCreatedAtBefore(statuses.pending().getName(), cutoff);
@@ -412,7 +435,27 @@ public class PaymentLifecycleStore {
                 .build();
     }
 
+    /**
+     * Applies a status transition, or declines it. Two things can resolve the same
+     * payment at once — an inbound callback and a status query — so the target is
+     * checked against what the payment already is: re-applying the same state is a
+     * no-op, and a less-informed verdict never overwrites a better one. The row is
+     * still written by the caller either way; only the status is held back.
+     */
     private PaymentInitiation updateStatus(PaymentInitiation initiation, Status status) {
+        if (!statuses.canTransition(initiation.getStatus(), status)) {
+            if (status.getName().equals(initiation.statusName())) {
+                // routine: re-asserting the state it already holds, e.g. an accepted
+                // response confirming a payment is still Pending
+                log.debug("[{}] initiation {} already {} — nothing to update",
+                        initiation.getProvider(), initiation.getId(), initiation.statusName());
+            } else {
+                // interesting: something tried to walk the payment backwards
+                log.warn("[{}] initiation {} is {} — refusing to overwrite with the less informed {}",
+                        initiation.getProvider(), initiation.getId(), initiation.statusName(), status.getName());
+            }
+            return initiation;
+        }
         initiation.setStatus(status);
         initiation.setUpdatedAt(LocalDateTime.now());
         return initiationRepository.save(initiation);

src/main/java/com/test/payment/service/StatusCatalog.java

@@ -30,6 +30,7 @@ public class StatusCatalog {
     private static final String SUCCESS = "Success";
     private static final String PENDING = "Pending";
     private static final String FAILED = "Failed";
+    private static final String UNRESOLVED = "Unresolved";
 
     /**
      * What {@code statuses} is seeded with when a state is missing. The table is the
@@ -39,7 +40,17 @@ public class StatusCatalog {
             new StatusDefinition(PAID, "Payment settled and confirmed — a provider receipt exists"),
             new StatusDefinition(SUCCESS, "Provider reported the collection succeeded"),
             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(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"));
+
+    /**
+     * Least to most informed. A payment may only move <em>up</em> this order, which is
+     * what stops a callback and a status query racing each other into a flapping
+     * 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<String> PRECEDENCE = List.of(PENDING, UNRESOLVED, FAILED, SUCCESS, PAID);
 
     private final StatusRepository statusRepository;
     private final Map<String, Status> byName = new ConcurrentHashMap<>();
@@ -68,6 +79,33 @@ public class StatusCatalog {
         return require(FAILED);
     }
 
+    /**
+     * Outcome unknown: pending for longer than the reconciliation window is willing to
+     * wait. Deliberately distinct from Failed — we are not claiming the payment failed,
+     * only that we never found out.
+     */
+    public Status unresolved() {
+        return require(UNRESOLVED);
+    }
+
+    /**
+     * 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,
+     * and a less-informed verdict can never overwrite a better one.
+     */
+    public boolean canTransition(Status current, Status target) {
+        if (target == null) {
+            return false;
+        }
+        return current == null || rank(target) > rank(current);
+    }
+
+    private int rank(Status status) {
+        int index = PRECEDENCE.indexOf(status.getName());
+        // an unknown state ranks lowest, so it never blocks a real verdict
+        return index < 0 ? -1 : index;
+    }
+
     /** True when the payment is still open — decides whether to re-query the provider. */
     public boolean isPending(String statusName) {
         return PENDING.equals(statusName);

src/main/resources/application.yml

@@ -82,8 +82,10 @@ springdoc:
 payments:
   token-expiry-buffer-seconds: 60
   reconciliation:
-    # transactions still PENDING after this age are re-checked against the provider
-    pending-age: 5m
+    # 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
     # how often the reconciliation job runs
     fixed-delay: 60s