Spring Data R2DBC Advanced Patterns and Stripe Integration: How @Transactional Reactive Proxy Re-runs the Method Body on Each .retryWhen() Re-subscription Generating UUID_B Inside the Transaction Boundary, Flux.merge() with Downstream retryWhen() Restarts All Customer Pipelines Including Already-Charged Customers, and Mono.cache() Defined Per Service Method Invocation Creates a New Memoization Scope on Each Retry Call

The prior R2DBC post in this series covered Kotlin-coroutine-specific failure modes: @Transactional suspend fun combined with @Retryable, flatMap placement of retryWhen(), and TransactionalOperator wrapping a UUID-generating Mono. This post shifts to pure Java Reactor patterns with Spring Data R2DBC. The three failure modes here are mechanically distinct from the coroutines post and from all earlier posts in the series. Each is rooted in a different aspect of how Project Reactor’s subscription model interacts with Spring’s reactive transaction proxy, Flux concurrency operators, and the Mono.cache() memoization operator.

The common thread is re-subscription. In Reactor, re-subscription is the mechanism behind retries: .retry(), .retryWhen(), and any operator that re-subscribes an upstream publisher all cause every cold publisher in the chain to re-execute. UUID.randomUUID() is a cold-publisher equivalent: it produces a new value each time it is evaluated. If UUID generation is anywhere in a reactive chain that can be re-subscribed, any re-subscription generates a new UUID. The three failure modes in this post differ in what causes the re-subscription: a reactive transaction proxy, a concurrent merge operator, and an incorrectly scoped memoization operator.

All three result in the same observable outcome: Stripe receives two requests for the same logical billing event, each with a distinct Idempotency-Key header, and creates ch_A and ch_B — two charges on the same customer for the same billing period.

Background: how Spring’s reactive @Transactional proxy works with R2dbcTransactionManager

Spring Framework 5.2 introduced reactive transaction support via ReactiveTransactionManager. The R2DBC variant is R2dbcTransactionManager. When you annotate a method that returns Mono<T> or Flux<T> with @Transactional, Spring generates a CGLIB proxy (or uses an interface proxy for components that implement interfaces). The proxy intercepts the method call via ReactiveTransactionInterceptor.

Critically, ReactiveTransactionInterceptor does not call the underlying method immediately when the proxy method is invoked. Instead, it returns a Mono.defer(() -> ...) that, on subscription, does the following: obtains a reactive transaction from the R2dbcTransactionManager, binds the transaction’s Connection to the Reactor context (via TransactionContextManager), invokes the actual method body to get its Mono, registers transaction-commit and rollback operators on that Mono, and subscribes. Each subscription to the proxy’s returned Mono runs this entire sequence from the top: new connection, new transaction, new method body invocation, new UUID.

This deferred-invocation model is correct and necessary: it allows transaction lifecycle to be managed in the Reactor context rather than in thread-local storage (which does not work across async operator boundaries). But it has a consequence that surprises developers: if a caller applies .retryWhen() to the Mono returned by calling the proxy method, the retry re-subscribes that Mono, and the Mono.defer() inside the proxy re-executes the method body. The original method body, including any UUID.randomUUID() call, runs again.

Mode 1: @Transactional reactive method returning Mono<Void> + .retryWhen() on the caller — the proxy re-runs the method body on each retry re-subscription, generating UUID_B inside the new transaction boundary

The developer writes a service class with a method annotated @Transactional that returns Mono<Void>. The method generates an idempotency key, saves a BillingAttempt row via R2dbcEntityTemplate, and calls Stripe. A billing scheduler invokes this method and applies .retryWhen(Retry.fixedDelay(3, Duration.ofSeconds(2))) to handle transient database connection timeouts from the connection pool.

// BillingService.java — unsafe: @Transactional reactive method + UUID at method entry
// Spring's reactive transaction proxy wraps this method in Mono.defer().
// .retryWhen() applied by the caller re-subscribes the proxy Mono.
// Re-subscription → Mono.defer() re-executes → method body runs again → UUID_B.

@Service
public class BillingService {

    private final R2dbcEntityTemplate template;
    private final StripeClient stripeClient;

    public BillingService(R2dbcEntityTemplate template, StripeClient stripeClient) {
        this.template = template;
        this.stripeClient = stripeClient;
    }

    // Developer's reasoning:
    //   "@Transactional manages the database connection via R2dbcTransactionManager.
    //    If the DB write fails, the transaction rolls back.
    //    The caller's .retryWhen() retries the Stripe call — UUID_A is generated once
    //    per billing event because billingCharge() is called once and the UUID is set
    //    at the top of the method before .retryWhen() gets involved."
    //
    // The problem:
    //   Spring's ReactiveTransactionInterceptor returns Mono.defer(() -> methodBody()).
    //   Each subscription to this Mono calls methodBody() — including UUID.randomUUID().
    //   When the caller applies .retryWhen() to the Mono returned by calling billingCharge(),
    //   .retryWhen() re-subscribes the upstream — the upstream IS the proxy Mono —
    //   the proxy Mono IS Mono.defer(() -> methodBody()) —
    //   Mono.defer() re-executes methodBody() — UUID.randomUUID() generates UUID_B.
    //   The new transaction commits UUID_B to the DB and sends ch_B to Stripe.

    @Transactional
    public Mono<Void> billingCharge(String customerId, long amountCents, String billingPeriod) {
        // UUID generated here — inside the @Transactional method body.
        // This line re-executes on every re-subscription to the proxy Mono.
        String idempotencyKey = UUID.randomUUID().toString();

        BillingAttempt attempt = BillingAttempt.pending(customerId, amountCents,
                billingPeriod, idempotencyKey);

        return template.insert(attempt)
                .then(Mono.fromCallable(() ->
                        stripeClient.createCharge(customerId, amountCents, idempotencyKey)))
                .then();
    }
}

// BillingScheduler.java — the caller applies .retryWhen() to handle transient DB errors.
// This re-subscribes billingService.billingCharge() on each retry attempt.
// Retry attempt 1: subscription #1 → methodBody() → UUID_A → ch_A committed at Stripe.
// If the DB write fails: @Transactional rolls back — DB has no BillingAttempt row.
// .retryWhen() sees the error — waits fixedDelay — re-subscribes.
// Retry attempt 2: subscription #2 → methodBody() → UUID_B → ch_B committed at Stripe.
// ch_A and ch_B both exist at Stripe. Customer charged twice.

@Component
public class BillingScheduler {

    private final BillingService billingService;

    public BillingScheduler(BillingService billingService) {
        this.billingService = billingService;
    }

    @Scheduled(fixedRate = 60_000)
    public void runBillingCycle() {
        // pendingBillingEvents() returns a Flux of customer billing requests.
        pendingBillingEvents()
                .flatMap(event ->
                        billingService.billingCharge(
                                        event.customerId(),
                                        event.amountCents(),
                                        event.billingPeriod())
                                // .retryWhen() applied to the Mono returned by billingCharge().
                                // This Mono IS the reactive @Transactional proxy Mono.
                                // Re-subscription → new @Transactional method invocation → UUID_B.
                                .retryWhen(Retry.fixedDelay(3, Duration.ofSeconds(2))
                                        .filter(ex -> ex instanceof R2dbcNonTransientException
                                                || ex instanceof ConnectionFactoryException))
                )
                .subscribe();
    }
}

The developer’s mental model assumes that billingService.billingCharge(...) is called once per billing event and returns a Mono<Void> that represents “one billing charge in progress.” In an imperative programming model, calling a method is the execution; returning a value is the result. The developer expects UUID.randomUUID() to have been evaluated at the call site, and the returned Mono to carry UUID_A as a captured value. But in reactive programming, calling a method that returns Mono assembles a pipeline description; execution happens on subscription. When the proxy intercepts the call, it wraps the original method in Mono.defer(() -> originalMethod()). The proxy call returns a cold Mono. Each subscription to that cold Mono calls originalMethod().

The distinction between “method call” and “Mono subscription” is invisible when there is exactly one subscriber and no retries. The distinction becomes catastrophic when .retryWhen() creates additional subscribers. The developer observes that .retryWhen() is chained on the Mono returned by the proxy method call, and expects that the retry scope starts after the method call — below the @Transactional boundary. This is the wrong mental model. In Reactor, operators are assembled into a pipeline. .retryWhen() is assembled into the pipeline at the point it appears. When it triggers, it re-subscribes the upstream of that point in the pipeline. The upstream of .retryWhen() in the scheduler code is billingService.billingCharge(...)’s returned Mono, which is the proxy’s cold Mono.defer(). Re-subscribing the proxy Mono re-runs the method.

This failure mode is distinct from the coroutines post’s @Transactional suspend fun + @Retryable combination. In the coroutines case, @Retryable is a Spring AOP interceptor that calls proceed() on the interceptor chain to re-invoke the method imperatively. In this Java Reactor case, there is no AOP retry interceptor; the retry is a Reactor operator (.retryWhen()) applied at the call site. The mechanism is different but the outcome is the same: the method body runs again with a new UUID.

Fixing Mode 1: compute the idempotency key outside the @Transactional method, pass it as a parameter

The idempotency key must be computed before the @Transactional boundary, so that all retry subscriptions use the same key. The cleanest fix is to move the key computation to the caller and pass it as a method parameter.

// BillingService.java — fixed: idempotency key is a method parameter, not generated internally
@Service
public class BillingService {

    private final R2dbcEntityTemplate template;
    private final StripeClient stripeClient;

    public BillingService(R2dbcEntityTemplate template, StripeClient stripeClient) {
        this.template = template;
        this.stripeClient = stripeClient;
    }

    @Transactional
    public Mono<Void> billingCharge(String customerId, long amountCents,
                                     String billingPeriod, String idempotencyKey) {
        // Idempotency key is a parameter — the caller controls its value.
        // All re-subscriptions to this proxy Mono re-run the method body,
        // but idempotencyKey is the same value passed from the caller.
        // The DB guard short-circuits if the first attempt already committed.

        return template.exists(
                        query(where("idempotency_key").is(idempotencyKey)),
                        BillingAttempt.class)
                .flatMap(alreadyProcessed -> {
                    if (alreadyProcessed) {
                        return Mono.empty(); // idempotent — already committed
                    }
                    BillingAttempt attempt = BillingAttempt.pending(customerId, amountCents,
                            billingPeriod, idempotencyKey);
                    return template.insert(attempt)
                            .then(Mono.fromCallable(() ->
                                    stripeClient.createCharge(customerId, amountCents, idempotencyKey)))
                            .then();
                });
    }
}

// BillingScheduler.java — fixed: content-hash key computed before the proxy call
@Component
public class BillingScheduler {

    private final BillingService billingService;

    public BillingScheduler(BillingService billingService) {
        this.billingService = billingService;
    }

    @Scheduled(fixedRate = 60_000)
    public void runBillingCycle() {
        pendingBillingEvents()
                .flatMap(event -> {
                    // Content-hash key computed here — before the @Transactional proxy call.
                    // Same event fields → same key → stable across all retryWhen() re-subscriptions.
                    String idempotencyKey = contentHashKey(
                            event.customerId(),
                            event.billingPeriod(),
                            event.amountCents());

                    return billingService.billingCharge(
                                    event.customerId(),
                                    event.amountCents(),
                                    event.billingPeriod(),
                                    idempotencyKey)
                            // .retryWhen() re-subscribes the proxy Mono — method body re-runs —
                            // but idempotencyKey is passed as a parameter and is already computed
                            // outside the lambda (captured from the flatMap body above).
                            // All retry attempts use the same idempotencyKey value.
                            .retryWhen(Retry.fixedDelay(3, Duration.ofSeconds(2))
                                    .filter(ex -> ex instanceof R2dbcNonTransientException));
                })
                .subscribe();
    }

    private String contentHashKey(String customerId, String billingPeriod, long amountCents) {
        String input = customerId + "|" + billingPeriod + "|" + amountCents;
        try {
            MessageDigest digest = MessageDigest.getInstance("SHA-256");
            byte[] hash = digest.digest(input.getBytes(StandardCharsets.UTF_8));
            return HexFormat.of().formatHex(hash).substring(0, 36);
        } catch (NoSuchAlgorithmException e) {
            throw new IllegalStateException("SHA-256 not available", e);
        }
    }
}

The key insight: the idempotencyKey variable is computed in the flatMap body of the scheduler, before the proxy method is called. It is captured by the lambda closure. Even when .retryWhen() re-subscribes the proxy Mono and the method body re-runs, the method receives the same idempotencyKey value passed as the argument. The Stripe call on the second attempt sends UUID_A again; Stripe returns the existing charge without creating ch_B.

The DB guard inside the @Transactional method provides a second layer of protection. On a retry where the first attempt fully succeeded (DB write committed, Stripe committed ch_A), the DB guard short-circuits before the Stripe call. This handles the edge case where the transaction committed but the downstream error was a network timeout reading the response — the caller saw an error, retried, but the first attempt had actually succeeded. Without the DB guard, the retry would call Stripe with UUID_A again (which Stripe safely deduplicates), but the redundant HTTP call is avoidable.

Mode 2: Flux.merge() of per-customer billing Monos + .retryWhen() applied to the outer merged Flux — merge re-subscribes all source publishers on retry, generating UUID_B for every customer including those already charged

The developer builds a batch billing pipeline that processes a list of customers in parallel. Instead of using Flux.flatMap() with a customer Flux as the source, the developer pre-builds one Mono<Void> per customer and merges them: Flux.merge(billingMonos). This pattern is sometimes used when each customer’s billing Mono requires distinct configuration that is easier to express at construction time rather than inside a flatMap body. A .retryWhen() operator is applied to the merged Flux to handle transient failures in any of the concurrent billing operations.

The failure is rooted in how Flux.merge() and .retryWhen() interact. Flux.merge(publisher1, publisher2, ...) subscribes to all source publishers immediately and concurrently when the merged Flux is subscribed. It does not process them sequentially. All source publishers are active at the same time. When any source publisher emits an error and the error propagates to the merged Flux, .retryWhen() re-subscribes the merged Flux. Re-subscribing the merged Flux causes merge() to re-subscribe all its source publishers simultaneously. Each source publisher is a per-customer billing Mono that contains UUID.randomUUID(). All of them re-execute. UUID_B is generated for every customer.

// BatchBillingService.java — unsafe: Flux.merge() + .retryWhen() outside merge
// merge() subscribes all source publishers immediately.
// retryWhen() re-subscribes the merged Flux on error.
// Re-subscription → merge() re-subscribes all source publishers → all UUIDs regenerated.

@Service
public class BatchBillingService {

    private final R2dbcEntityTemplate template;
    private final StripeClient stripeClient;

    public BatchBillingService(R2dbcEntityTemplate template, StripeClient stripeClient) {
        this.template = template;
        this.stripeClient = stripeClient;
    }

    public Mono<Void> runBillingBatch(List<BillingEvent> events) {
        // Build one billing Mono per customer — UUID generated inside each Mono's lazy chain.
        // At this point, none of the Monos have been subscribed.
        // UUID.randomUUID() has NOT been called yet.
        List<Mono<Void>> billingMonos = events.stream()
                .map(this::buildBillingMono)
                .toList();

        // Developer's reasoning:
        //   "Flux.merge() runs all customer billing operations concurrently.
        //    .retryWhen() retries any that fails.
        //    Customers that succeeded in the first subscription are already done —
        //    their Mono completed — retryWhen() only re-triggers the failed one."
        //
        // The problem:
        //   retryWhen() re-subscribes the MERGED Flux — not individual failed Monos.
        //   Flux.merge(billingMonos) treated as a single cold publisher from retryWhen's perspective.
        //   Re-subscription → merge() subscribes to ALL billingMonos again.
        //   ALL buildBillingMono() lambdas re-execute.
        //   UUID.randomUUID() generates UUID_B for every customer.
        //   stripeClient.createCharge(UUID_B) → ch_B for every customer.
        //   Customers already charged (ch_A): now also charged via ch_B.

        return Flux.merge(billingMonos)
                .retryWhen(Retry.fixedDelay(3, Duration.ofSeconds(1))
                        .filter(ex -> ex instanceof R2dbcTransientException))
                .then();
    }

    private Mono<Void> buildBillingMono(BillingEvent event) {
        // Mono.defer() ensures UUID is generated on subscription, not at construction.
        // This is the correct pattern for a single non-retried subscription.
        // But it becomes problematic when retryWhen() outside merge() re-subscribes all Monos.
        return Mono.defer(() -> {
            // UUID generated here — at subscription time (Mono.defer semantics).
            // On re-subscription from retryWhen(): this lambda re-executes → UUID_B.
            String idempotencyKey = UUID.randomUUID().toString();

            BillingAttempt attempt = BillingAttempt.pending(
                    event.customerId(), event.amountCents(),
                    event.billingPeriod(), idempotencyKey);

            return template.insert(attempt)
                    .then(Mono.fromCallable(() ->
                            stripeClient.createCharge(
                                    event.customerId(),
                                    event.amountCents(),
                                    idempotencyKey)))
                    .then();
        });
    }
}

The subtle aspect of this failure is the developer’s model of retryWhen() as a per-element operator. In a Flux.flatMap() chain, the developer might correctly observe that placing .retryWhen() inside the flatMap lambda (per-element) retries only the failing element, while placing it outside (on the outer Flux) retries the entire outer Flux. But with Flux.merge(), the concept of “the failing element” does not exist in the same way: the merge operator does not expose individual element retry boundaries. The merged Flux is a single reactive sequence that interleaves emissions from all sources. When the merged Flux terminates with an error, .retryWhen() re-subscribes the merged Flux as a unit, which causes all sources to restart.

This is mechanically distinct from the coroutines post’s Flux.flatMap().retryWhen() placement failure, even though the surface-level symptom (all customers re-charged) is similar. In the coroutines case, the root cause is flatMap operator placement and the sequential restart of a Flux source. In this case, the root cause is merge()’s all-upfront subscription model: merge() does not distinguish between “completed sources” and “failed sources” from retryWhen()’s perspective — it restarts all of them. A Flux.flatMap() pipeline over a customer Flux with per-element retryWhen() inside the flatMap body would correctly retry only the failing customer; Flux.merge() of pre-built Monos with outer retryWhen() cannot achieve this because the sources are not individually addressable from outside the merge.

Fixing Mode 2: content-hash key pre-computed at batch construction time, DB guard inside each billing Mono

The fix has two parts. First, move UUID generation out of Mono.defer() and compute a content-hash key at batch construction time (when the list of BillingEvents is processed). The key is now a value captured by the lambda closure, not a randomUUID() call inside the deferred block. Second, if any outer retry is needed at all, consider whether it should be on the individual billing Mono (per-element, inside the pre-built Mono) rather than on the merged Flux.

// BatchBillingService.java — fixed: content-hash keys computed at construction time
@Service
public class BatchBillingService {

    private final R2dbcEntityTemplate template;
    private final StripeClient stripeClient;

    public BatchBillingService(R2dbcEntityTemplate template, StripeClient stripeClient) {
        this.template = template;
        this.stripeClient = stripeClient;
    }

    public Mono<Void> runBillingBatch(List<BillingEvent> events) {
        List<Mono<Void>> billingMonos = events.stream()
                .map(event -> {
                    // Content-hash key computed HERE — at batch construction, not at subscription.
                    // Same event fields → same key → all re-subscriptions of this Mono
                    // (from per-element retryWhen inside buildBillingMono) use the same key.
                    String idempotencyKey = contentHashKey(
                            event.customerId(),
                            event.billingPeriod(),
                            event.amountCents());
                    return buildBillingMono(event, idempotencyKey);
                })
                .toList();

        // No retryWhen() at the merge level.
        // Per-element retry is inside buildBillingMono().
        // Each billing Mono retries independently with its own stable idempotencyKey.
        return Flux.merge(billingMonos).then();
    }

    private Mono<Void> buildBillingMono(BillingEvent event, String idempotencyKey) {
        // idempotencyKey is a parameter — stable value captured from the outer scope.
        // No UUID.randomUUID() inside this method.
        return template.exists(
                        query(where("idempotency_key").is(idempotencyKey)),
                        BillingAttempt.class)
                .flatMap(alreadyProcessed -> {
                    if (alreadyProcessed) {
                        return Mono.<Void>empty(); // idempotent
                    }
                    BillingAttempt attempt = BillingAttempt.pending(
                            event.customerId(), event.amountCents(),
                            event.billingPeriod(), idempotencyKey);
                    return template.insert(attempt)
                            .then(Mono.fromCallable(() ->
                                    stripeClient.createCharge(
                                            event.customerId(),
                                            event.amountCents(),
                                            idempotencyKey)))
                            .then();
                })
                // Per-element retry with stable idempotencyKey.
                // Re-subscriptions of this Mono re-run the method body,
                // but idempotencyKey is passed as a parameter and does not change.
                .retryWhen(Retry.fixedDelay(2, Duration.ofSeconds(1))
                        .filter(ex -> ex instanceof R2dbcTransientException));
    }

    private String contentHashKey(String customerId, String billingPeriod, long amountCents) {
        String input = customerId + "|" + billingPeriod + "|" + amountCents;
        try {
            MessageDigest digest = MessageDigest.getInstance("SHA-256");
            byte[] hash = digest.digest(input.getBytes(StandardCharsets.UTF_8));
            return HexFormat.of().formatHex(hash).substring(0, 36);
        } catch (NoSuchAlgorithmException e) {
            throw new IllegalStateException("SHA-256 not available", e);
        }
    }
}

In the fixed version, each per-customer Mono has its own .retryWhen() applied inside buildBillingMono(). This .retryWhen() retries the individual customer’s billing chain without affecting other customers. The outer Flux.merge() no longer carries a .retryWhen(). A customer that fails retries independently; a customer that succeeds is not touched. The idempotencyKey passed into buildBillingMono() is a stable content-hash key; all retry re-subscriptions of that customer’s Mono use the same key.

An alternative fix, if moving retry inside each Mono is impractical, is to move UUID generation entirely outside the Mono.defer() scope and ensure the key is captured as a final value before merge(). This is what the stream map in the fixed runBillingBatch above accomplishes: by the time Flux.merge(billingMonos) is called, all idempotency keys are already computed and captured in each Mono’s closure as stable strings. Even if an outer .retryWhen() were added back to the merged Flux, the content-hash keys would ensure all re-subscriptions use the same keys per customer, and the DB guard would short-circuit customers already processed.

Mode 3: Mono.cache() defined inside a service method body — developer expects cache() to memoize UUID_A for the duration of a billing operation, but each service method invocation creates a new Mono.cache() instance that memoizes a freshly generated UUID_B

The third failure mode is subtler than the first two because it involves a Reactor operator whose purpose is explicitly to prevent redundant re-evaluation: Mono.cache(). The developer has read the documentation, understands that Mono.cache() memoizes the first emitted item and returns the memoized item to all subsequent subscribers, and applies this to the idempotency key generation to prevent UUID re-generation on re-subscription. The reasoning is correct about how cache() works. The error is about scope: the Mono.cache() instance that memoizes UUID_A is not the same instance that is subscribed on the retry path.

// BillingService.java — unsafe Mode 3: Mono.cache() defined inside service method body
// cache() memoizes the UUID_A on first subscription to THIS Mono instance.
// But: when the caller retries by re-calling billingCharge(), a NEW Mono.cache() instance
// is created. This new instance has no memoized value.
// On subscription, its Mono.fromCallable() runs → UUID.randomUUID() → UUID_B.

@Service
public class BillingService {

    private final R2dbcEntityTemplate template;
    private final StripeClient stripeClient;

    public BillingService(R2dbcEntityTemplate template, StripeClient stripeClient) {
        this.template = template;
        this.stripeClient = stripeClient;
    }

    // Developer's reasoning:
    //   "Mono.fromCallable(() -> UUID.randomUUID().toString()).cache() creates a cached Mono.
    //    The first subscription evaluates the callable and memoizes UUID_A.
    //    Any re-subscription to this Mono — including on retry — returns UUID_A from the cache.
    //    This is exactly what I need: a stable UUID for the duration of this billing operation."
    //
    // The problem:
    //   keyMono is a local variable inside billingCharge().
    //   A new Mono.cache() instance is created on every call to billingCharge().
    //   If the caller retries by calling billingCharge() again (not by re-subscribing
    //   to the same keyMono), the new call creates a new Mono.cache() instance.
    //   The new instance has no memoized value.
    //   The first subscription to the new instance evaluates the callable → UUID_B.
    //   Stripe receives UUID_B → ch_B.

    @Transactional
    public Mono<Void> billingCharge(String customerId, long amountCents, String billingPeriod) {
        // NEW Mono.cache() instance created on each call to billingCharge().
        // Memoizes UUID_A only within the lifecycle of this specific instance.
        // A retry that calls billingCharge() again creates a DIFFERENT instance.
        Mono<String> keyMono = Mono.fromCallable(() -> UUID.randomUUID().toString()).cache();

        // On first subscription: callable runs → UUID_A memoized in keyMono.
        // If .retryWhen() is applied to the chain INSIDE this method (not outside the proxy):
        //   Re-subscription re-uses the same keyMono instance → cache returns UUID_A → correct.
        // If the CALLER retries by calling billingCharge() again:
        //   New billingCharge() call → new keyMono instance → UUID_B.
        return keyMono
                .flatMap(idempotencyKey -> {
                    BillingAttempt attempt = BillingAttempt.pending(customerId, amountCents,
                            billingPeriod, idempotencyKey);
                    return template.insert(attempt)
                            .then(Mono.fromCallable(() ->
                                    stripeClient.createCharge(customerId, amountCents,
                                            idempotencyKey)))
                            .then();
                });
    }
}

// CallerService.java — the retry calls billingCharge() again, not re-subscribed the same Mono.
// Each call to billingCharge() creates a new Mono.cache() instance.
// keyMono from call #1 is not accessible here. UUID_B on call #2.

@Service
public class CallerService {

    private final BillingService billingService;

    public CallerService(BillingService billingService) {
        this.billingService = billingService;
    }

    public Mono<Void> processWithRetry(BillingEvent event) {
        // This pattern calls billingCharge() up to 4 times (1 initial + 3 retries).
        // Each call creates a new Mono.cache() instance inside billingCharge().
        // Attempt 1: billingCharge() call #1 → keyMono #1 → UUID_A → ch_A committed.
        //   If DB write fails: transaction rolled back. Stripe committed ch_A.
        // Attempt 2 (retry): billingCharge() call #2 → keyMono #2 → UUID_B → ch_B committed.
        //   Customer charged twice.

        return Mono.defer(() -> billingService.billingCharge(
                        event.customerId(),
                        event.amountCents(),
                        event.billingPeriod()))
                .retryWhen(Retry.fixedDelay(3, Duration.ofSeconds(2)));

        // Mono.defer() wraps the billingCharge() call itself — this is necessary to call
        // billingCharge() on each re-subscription (retryWhen re-subscribes the defer).
        // But it means billingCharge() is called again on each retry:
        // new call → new Mono.cache() → UUID_B.
    }
}

There is a nuanced correctness boundary for Mono.cache() in this context. If the .retryWhen() were applied inside the billingCharge() method, on the chain that includes keyMono, the cache() would correctly memoize UUID_A across all internal retry re-subscriptions, because the same keyMono instance is referenced by the lambda closure:

// This INTERNAL pattern is correct — retryWhen re-subscribes the flatMap chain,
// but keyMono is the same instance, and cache() returns the memoized UUID_A.
@Transactional
public Mono<Void> billingCharge(String customerId, long amountCents, String billingPeriod) {
    Mono<String> keyMono = Mono.fromCallable(() -> UUID.randomUUID().toString()).cache();

    return keyMono
            .flatMap(idempotencyKey -> {
                BillingAttempt attempt = BillingAttempt.pending(customerId, amountCents,
                        billingPeriod, idempotencyKey);
                return template.insert(attempt)
                        .then(Mono.fromCallable(() ->
                                stripeClient.createCharge(customerId, amountCents, idempotencyKey)))
                        .then();
            })
            // retryWhen INSIDE the method — re-subscribes the flatMap chain,
            // but keyMono instance is the same → cache() returns memoized UUID_A.
            .retryWhen(Retry.fixedDelay(3, Duration.ofSeconds(2)));
}

This internal-retry version is correct for the memoization purpose: the same keyMono instance is closed over by the lambda, and all re-subscriptions from .retryWhen() go through the same cache() instance, which returns the memoized UUID_A. However, it is still problematic for a different reason: the reactive @Transactional proxy wraps the method in Mono.defer(), and if anything outside this method re-subscribes the proxy Mono, the entire method body runs again (Mode 1 failure). The internal keyMono.cache() does not help in that case because a new method call creates a new keyMono instance.

The deeper problem with Mono.cache() for idempotency key management is that it creates a coupling between the memoization lifetime and the subscription scope. The developer wants the UUID to be stable for “this billing operation,” but “this billing operation” is not a concept that Reactor formalizes. It has to be implemented explicitly. Mono.cache() formalizes “this Mono instance” as the scope, not “this billing operation.” The two coincide when the Mono instance outlives all retry paths, but diverge whenever a retry path creates a new Mono instance (via Mono.defer(), a fresh method call, or any operator that creates a new upstream publisher).

Fixing Mode 3: replace Mono.cache() with a content-hash key computed eagerly before the reactive chain

The fix eliminates the dependency on instance scope entirely. A content-hash key derived from the billing event’s immutable fields is stable by construction: the same inputs always produce the same key, regardless of which instance, which subscription, or which retry attempt evaluates the hash function.

// BillingService.java — fixed Mode 3: content-hash key replaces Mono.cache()
@Service
public class BillingService {

    private final R2dbcEntityTemplate template;
    private final StripeClient stripeClient;

    public BillingService(R2dbcEntityTemplate template, StripeClient stripeClient) {
        this.template = template;
        this.stripeClient = stripeClient;
    }

    @Transactional
    public Mono<Void> billingCharge(String customerId, long amountCents,
                                     String billingPeriod, String idempotencyKey) {
        // idempotencyKey is a parameter — no UUID generation inside this method.
        // Caller computes content-hash key from the billing event's immutable fields.
        // All re-subscriptions to this proxy Mono (from retryWhen at the call site)
        // receive the same idempotencyKey value as the method parameter.

        return template.exists(
                        query(where("idempotency_key").is(idempotencyKey)),
                        BillingAttempt.class)
                .flatMap(alreadyProcessed -> {
                    if (alreadyProcessed) {
                        return Mono.<Void>empty();
                    }
                    BillingAttempt attempt = BillingAttempt.pending(customerId, amountCents,
                            billingPeriod, idempotencyKey);
                    return template.insert(attempt)
                            .then(Mono.fromCallable(() ->
                                    stripeClient.createCharge(customerId, amountCents,
                                            idempotencyKey)))
                            .then();
                });
    }
}

// CallerService.java — fixed: content-hash key computed before Mono.defer()
@Service
public class CallerService {

    private final BillingService billingService;

    public CallerService(BillingService billingService) {
        this.billingService = billingService;
    }

    public Mono<Void> processWithRetry(BillingEvent event) {
        // Content-hash key computed once, before Mono.defer().
        // Captured by the Mono.defer() lambda — stable across all re-subscriptions.
        // Even if the @Transactional proxy re-runs the method body on retry,
        // the method receives the same idempotencyKey as a parameter.
        final String idempotencyKey = contentHashKey(
                event.customerId(),
                event.billingPeriod(),
                event.amountCents());

        return Mono.defer(() -> billingService.billingCharge(
                        event.customerId(),
                        event.amountCents(),
                        event.billingPeriod(),
                        idempotencyKey))
                .retryWhen(Retry.fixedDelay(3, Duration.ofSeconds(2)));
    }

    private String contentHashKey(String customerId, String billingPeriod, long amountCents) {
        String input = customerId + "|" + billingPeriod + "|" + amountCents;
        try {
            MessageDigest digest = MessageDigest.getInstance("SHA-256");
            byte[] hash = digest.digest(input.getBytes(StandardCharsets.UTF_8));
            return HexFormat.of().formatHex(hash).substring(0, 36);
        } catch (NoSuchAlgorithmException e) {
            throw new IllegalStateException("SHA-256 not available", e);
        }
    }
}

The content-hash key needs no memoization because it is idempotent: computing it twice with the same inputs produces the same string. Mono.cache() was providing memoization to avoid re-calling UUID.randomUUID(); a content-hash key never needs memoization because the hash function itself is the “cache.” The same inputs always produce the same output. The developer can discard the Mono.cache() approach entirely and replace it with the simpler, more correct content-hash pattern.

If the developer specifically wants to use Mono.cache() (for example, to avoid re-hashing on multiple subscriptions of the same downstream chain within a single method), the correct approach is to compute the hash key first and then wrap only the side-effectful downstream Mono in a Mono.cache() — not the UUID generation step. Caching a deterministic function’s output is never necessary; caching a side-effectful operation (like a database insert or a Stripe call) is occasionally useful for replay-safety. But in practice, the DB guard (checking for an existing BillingAttempt row by idempotency key) is a more reliable replay-safety mechanism than Mono.cache(), because the DB persists across process restarts while an in-memory Mono instance does not.

Comparison table: three Spring Data R2DBC advanced-pattern Stripe idempotency failure modes

Integration tests for all three modes with @DataR2dbcTest and WireMock

The integration test for Mode 1 must verify that .retryWhen() applied on the caller side does not generate UUID_B. It needs to be able to inspect the idempotency key sent to Stripe on each attempt. WireMock’s request journal provides this.

// BillingServiceRetryTest.java — Mode 1 integration test
// Tests that .retryWhen() on the caller does NOT cause UUID_B when key is content-hash derived.

@SpringBootTest
@AutoConfigureWireMock(port = 8089)
class BillingServiceRetryTest {

    @Autowired BillingService billingService;
    @Autowired BillingAttemptRepository billingAttemptRepository;

    @BeforeEach
    void setUp() {
        billingAttemptRepository.deleteAll().block();
        WireMock.reset();
    }

    @Test
    void retryWhen_doesNotGenerateUUID_B_whenKeyIsContentHash() {
        // Stripe returns 503 on first call, 200 on second — simulating transient failure.
        // With content-hash key: both attempts send the same Idempotency-Key header.
        stubFor(post(urlEqualTo("/v1/payment_intents"))
                .inScenario("stripe-retry")
                .whenScenarioStateIs(STARTED)
                .willReturn(serviceUnavailable())
                .willSetStateTo("second-attempt"));

        stubFor(post(urlEqualTo("/v1/payment_intents"))
                .inScenario("stripe-retry")
                .whenScenarioStateIs("second-attempt")
                .willReturn(okJson("{\"id\":\"pi_test\",\"status\":\"succeeded\"}")));

        String customerId = "cust_mode1";
        String billingPeriod = "2026-10";
        long amountCents = 4900L;
        String expectedKey = contentHashKey(customerId, billingPeriod, amountCents);

        // Run with retry at the caller level.
        Mono.defer(() -> billingService.billingCharge(customerId, amountCents,
                        billingPeriod, expectedKey))
                .retryWhen(Retry.fixedDelay(3, Duration.ZERO))
                .block(Duration.ofSeconds(5));

        // Assert: exactly two WireMock requests (one 503, one 200).
        List<LoggedRequest> stripeRequests = WireMock.findAll(
                postRequestedFor(urlEqualTo("/v1/payment_intents")));
        assertThat(stripeRequests).hasSize(2);

        // Assert: BOTH attempts sent the SAME Idempotency-Key.
        // This is the critical assertion — UUID_B would produce a different value.
        String keyAttempt1 = stripeRequests.get(0).getHeader("Idempotency-Key");
        String keyAttempt2 = stripeRequests.get(1).getHeader("Idempotency-Key");
        assertThat(keyAttempt1)
                .as("Retry must send the same Idempotency-Key as the first attempt")
                .isEqualTo(keyAttempt2);
        assertThat(keyAttempt1).isEqualTo(expectedKey);

        // Assert: exactly one BillingAttempt row.
        assertThat(billingAttemptRepository.count().block()).isEqualTo(1L);
    }
}

// BatchBillingRetryTest.java — Mode 2: Flux.merge() with per-element retryWhen()
// Tests that only the failing customer triggers a retry — not all customers.

@SpringBootTest
@AutoConfigureWireMock(port = 8089)
class BatchBillingRetryTest {

    @Autowired BatchBillingService batchBillingService;
    @Autowired BillingAttemptRepository billingAttemptRepository;

    @BeforeEach
    void setUp() {
        billingAttemptRepository.deleteAll().block();
        WireMock.reset();
    }

    @Test
    void perElementRetry_doesNotRestartSuccessfulCustomers() {
        // cust_A: Stripe returns 200 immediately.
        stubFor(post(urlEqualTo("/v1/payment_intents"))
                .withHeader("X-Customer-Id", equalTo("cust_A"))
                .willReturn(okJson("{\"id\":\"pi_A\",\"status\":\"succeeded\"}")));

        // cust_B: Stripe returns 503 on first attempt, 200 on second.
        stubFor(post(urlEqualTo("/v1/payment_intents"))
                .withHeader("X-Customer-Id", equalTo("cust_B"))
                .inScenario("cust-b-retry")
                .whenScenarioStateIs(STARTED)
                .willReturn(serviceUnavailable())
                .willSetStateTo("second"));

        stubFor(post(urlEqualTo("/v1/payment_intents"))
                .withHeader("X-Customer-Id", equalTo("cust_B"))
                .inScenario("cust-b-retry")
                .whenScenarioStateIs("second")
                .willReturn(okJson("{\"id\":\"pi_B\",\"status\":\"succeeded\"}")));

        List<BillingEvent> events = List.of(
                new BillingEvent("cust_A", 1000L, "2026-10"),
                new BillingEvent("cust_B", 2000L, "2026-10"));

        batchBillingService.runBillingBatch(events).block(Duration.ofSeconds(5));

        // Assert: cust_A was charged exactly ONCE.
        // If Mode 2 failure occurred, cust_A would have been re-charged when cust_B triggered retry.
        List<LoggedRequest> requestsForA = WireMock.findAll(
                postRequestedFor(urlEqualTo("/v1/payment_intents"))
                        .withHeader("X-Customer-Id", equalTo("cust_A")));
        assertThat(requestsForA)
                .as("cust_A must be charged exactly once — per-element retry must not restart it")
                .hasSize(1);

        // Assert: cust_B was charged exactly twice (one failure + one success).
        List<LoggedRequest> requestsForB = WireMock.findAll(
                postRequestedFor(urlEqualTo("/v1/payment_intents"))
                        .withHeader("X-Customer-Id", equalTo("cust_B")));
        assertThat(requestsForB).hasSize(2);

        // Assert: both cust_B attempts used the SAME Idempotency-Key (content-hash stable).
        String keyBAttempt1 = requestsForB.get(0).getHeader("Idempotency-Key");
        String keyBAttempt2 = requestsForB.get(1).getHeader("Idempotency-Key");
        assertThat(keyBAttempt1).isEqualTo(keyBAttempt2);

        // Assert: two BillingAttempt rows total (one per customer).
        assertThat(billingAttemptRepository.count().block()).isEqualTo(2L);
    }
}

// MonoCacheIdempotencyTest.java — Mode 3: verifies content-hash replaces Mono.cache()
// Tests that re-calling the service method does not generate a different idempotency key.

@SpringBootTest
@AutoConfigureWireMock(port = 8089)
class MonoCacheIdempotencyTest {

    @Autowired BillingService billingService;
    @Autowired BillingAttemptRepository billingAttemptRepository;

    @BeforeEach
    void setUp() {
        billingAttemptRepository.deleteAll().block();
        WireMock.reset();
        stubFor(post(urlEqualTo("/v1/payment_intents"))
                .willReturn(okJson("{\"id\":\"pi_test\",\"status\":\"succeeded\"}")));
    }

    @Test
    void callingServiceMethodTwice_usesSameIdempotencyKey() {
        String customerId = "cust_mode3";
        String billingPeriod = "2026-10";
        long amountCents = 9900L;
        String expectedKey = contentHashKey(customerId, billingPeriod, amountCents);

        // Call #1: simulates first attempt.
        billingService.billingCharge(customerId, amountCents, billingPeriod, expectedKey)
                .block(Duration.ofSeconds(5));

        // Reset DB to simulate the first transaction rolled back (no row committed).
        billingAttemptRepository.deleteAll().block();

        // Call #2: simulates retry (caller calls the method again, not re-subscribes same Mono).
        billingService.billingCharge(customerId, amountCents, billingPeriod, expectedKey)
                .block(Duration.ofSeconds(5));

        // Assert: both calls sent the SAME Idempotency-Key to Stripe.
        // With Mono.cache() per-invocation approach (broken), call #2 would send UUID_B.
        List<LoggedRequest> stripeRequests = WireMock.findAll(
                postRequestedFor(urlEqualTo("/v1/payment_intents")));
        assertThat(stripeRequests).hasSize(2);

        String keyCall1 = stripeRequests.get(0).getHeader("Idempotency-Key");
        String keyCall2 = stripeRequests.get(1).getHeader("Idempotency-Key");
        assertThat(keyCall1)
                .as("Second service method call must send the same Idempotency-Key as the first")
                .isEqualTo(keyCall2);
        assertThat(keyCall1).isEqualTo(expectedKey);
    }
}

Required assertions checklist for Spring Data R2DBC Stripe integration tests

1. Retry key stability. Apply .retryWhen(Retry.fixedDelay(1, Duration.ZERO)) at the same level as production retry. WireMock: return 503 on first request, 200 on second. Assert keyAttempt1.equals(keyAttempt2) via WireMock request journal.
2. Per-customer isolation in batch pipelines. Simulate one customer failing and one succeeding. Assert that the successful customer receives exactly one Stripe request (per-element retry does not restart already-successful customers).
3. Service method re-call stability. Call the service method twice with the same billing event parameters. Assert both calls send the same Idempotency-Key header. This catches Mono.cache()-per-invocation bugs and other per-instance UUID generation patterns.
4. DB guard idempotency. Pre-insert a BillingAttempt with the expected content-hash key. Call the service method. Assert Stripe was not called (DB guard short-circuited). Assert DB has exactly one row (no duplicate insert).
5. Content-hash determinism. Call contentHashKey(customerId, billingPeriod, amountCents) twice with identical inputs. Assert both calls return the same string. Call with a different amountCents. Assert the result differs.

Detection signals in production: identifying the specific R2DBC advanced-pattern failure mode

All three modes produce the same top-level observable: two Stripe charges for the same customer and billing period within a short window. The distinguishing signals help identify which mode is responsible.

Mode 1 signals (@Transactional proxy re-run on retryWhen()):

• The two Stripe charges are separated by the retryWhen() delay configured in the scheduler. For Retry.fixedDelay(3, Duration.ofSeconds(2)), the two charges are approximately 2–6 seconds apart (delay × attempt count).
• Application DB has a BillingAttempt row for UUID_B but not UUID_A: the first transaction rolled back (DB write failed after Stripe call), leaving only the second attempt’s row. Stripe Dashboard shows ch_A created before ch_B.
• If logging is enabled on the @Transactional proxy (Spring’s TransactionSynchronizationManager logging at DEBUG level), you see two “Creating new transaction” log entries for the same customer in the retry window.

Mode 2 signals (Flux.merge() + outer retryWhen()):

• Every customer in the batch has a duplicate Stripe charge. The Mode 1 and Mode 3 failures typically affect one customer per incident. Mode 2 produces simultaneous duplicates across all customers in the batch.
• All duplicate charges appear within the same retry window (the retryWhen() delay). The batch size is a multiplier: a 50-customer batch with Mode 2 produces up to 50 double-charge incidents simultaneously.
• If any customer has a DB guard that successfully deduplicated the second charge, that customer’s Stripe call count in the audit log is 2 (one per subscription) but Stripe shows only one charge (the second was deduplicated by Stripe because the key happened to be a content-hash). In a random UUID mode without the DB guard, all customers show 2 Stripe charges each.

Mode 3 signals (Mono.cache() per invocation):

• The two Stripe charges correspond to two separate calls to the service method (not to two subscriptions of the same Mono). Application logs show two distinct invocations of billingCharge() for the same customer and billing period.
• Unlike Mode 1 (which produces no UUID_A DB row), Mode 3 may produce two DB rows: one for UUID_A (if the first call’s transaction committed) and one for UUID_B (if the second call’s transaction committed). Both rows have the same customer_id and billing_period; only the idempotency_key differs. A unique constraint on (customer_id, billing_period) would catch this at the DB level, but developers often omit this constraint if they consider idempotency_key to be the primary deduplication key.
• A Keybrake proxy audit log query for SELECT * FROM audit_log WHERE customer_id = ? AND billing_period = ? ORDER BY created_at shows two entries with different idempotency_key values and timestamps separated by the caller’s retry interval (typically minutes for a scheduled retry vs. seconds for a reactive retryWhen() retry).

All three modes are auditable via Keybrake’s per-vendor spend cap. A spend-cap policy of daily_usd_cap: 10000 with an alert on “same customer charged twice within 60 seconds” catches all three modes. Mode 2 would additionally trigger an alert for every customer in the affected batch simultaneously — a cluster of alerts within the same minute is a strong indicator of Mode 2 (outer-merge retry) rather than Mode 1 or Mode 3 (single-customer retry).