Keybrake

Ratpack and Stripe Integration: Restricted API Keys, Spend Caps, and Agent Governance

By Keybrake ·

Ratpack’s Promise<T> is a lazy description of a computation — each subscription triggers a fresh execution of the producer chain from the factory lambda through every flatMap() and Blocking.get() step. Promise.retry() works by re-subscribing the upstream Promise on each failure: the entire chain re-executes on every retry attempt. UUID.randomUUID() inside a Blocking.get() supplier is evaluated at supplier-run time, not at Promise-construction time. The initial subscription runs the supplier, evaluates UUID_A, commits ch_A to Stripe before a socket timeout wraps into a StripeException; Promise.retry() re-subscribes, the supplier runs again, UUID_B is evaluated, and Stripe creates ch_B. A subtler variant arises from flatMap composition: when an outer Promise chain containing a flatMap lambda is wrapped in .retry(), the entire flatMap lambda body re-executes on each outer re-subscription — including any UUID.randomUUID() call that a developer placed outside the inner Blocking.get() supplier, believing it to be evaluated once per customer, not realising it is still inside the outer retry boundary. And ExecController.executor().scheduleAtFixedRate() — a raw ScheduledExecutorService — fires independently on every Ratpack node in a Kubernetes replicas:3 Deployment with no built-in cluster-wide coordination, each node generating its own UUID per customer and producing ch_A, ch_B, and ch_C per customer per billing period.

This post covers all three failure modes with Java code (Ratpack 1.9.x), content-hash stable keys captured before the Promise chain is assembled, the distinction between Promise-construction time and subscription time for UUID.randomUUID() placement, pg_try_advisory_lock() for cluster-wide scheduled billing serialization, and pre-flight PostgreSQL ON CONFLICT DO NOTHING as a durable billing mutex — plus per-billing-period vault keys via a spend-cap proxy as a hard financial backstop. For Spring Retry’s @Retryable AOP proxy re-invocation pattern, see the Spring Retry and Stripe Integration post. For Vert.x Web Client’s retryWhen() re-subscription semantics, see the Vert.x and Stripe Integration post. Ratpack’s Promise subscription model makes the failure boundaries structurally different from both: the retry boundary is the full Promise chain from construction point to the .retry() call, not just an annotated method body or a reactive operator.

Failure mode 1: Promise.retry() re-subscribes the upstream Promise chain — Blocking.get() supplier re-called per re-subscription — UUID.randomUUID() inside the blocking supplier evaluates fresh per retry attempt — initial subscription creates ch_A before StripeException — first re-subscription creates ch_B

In Ratpack, Blocking.get(Factory<T> factory) accepts a supplier-style lambda that runs on Ratpack’s blocking thread pool. Calling Blocking.get() constructs a Promise<T> that, when subscribed, dispatches the factory lambda to a blocking thread for execution. The Promise is lazy: the factory lambda does not run at the time Blocking.get() is called. It runs at the time the returned Promise is subscribed.

When .retry(n, duration) is chained onto the Promise returned by Blocking.get(), Ratpack’s retry implementation re-subscribes the upstream Promise on each failure. Re-subscribing the Blocking.get() Promise re-dispatches the factory lambda to the blocking thread pool. The factory lambda body executes from start to finish again, including every expression within it: variable declarations, object constructions, and calls to UUID.randomUUID(). There is no mechanism by which the retry infrastructure preserves or replays the values computed during the first subscription:

// UNSAFE: UUID.randomUUID() inside a Blocking.get() supplier combined with .retry().
// Blocking.get(factory) constructs a lazy Promise — factory is not called at construction time.
// Each Promise subscription dispatches the factory to a blocking thread.
// Promise.retry() re-subscribes the upstream on failure — re-dispatches the factory.
// UUID.randomUUID() inside the factory evaluates fresh on each dispatch.

import com.stripe.model.Charge;
import com.stripe.net.RequestOptions;
import com.stripe.param.ChargeCreateParams;
import ratpack.exec.Blocking;
import ratpack.exec.Promise;

import java.time.Duration;
import java.util.UUID;

public class BillingHandler {

    public Promise<Charge> chargeCustomer(String customerId, long amountCents,
                                           String billingPeriod) {

        return Blocking.get(() -> {
            // UNSAFE: Blocking.get() factory lambda re-called on each Promise re-subscription.
            // Promise.retry() below re-subscribes on StripeException — factory re-runs.
            // UUID.randomUUID() is a call expression evaluated each time the factory runs.
            //
            // Subscription 1 (initial attempt):
            //   UUID = "7a3f1b2c-4d5e-6f7a-8b9c-0d1e2f3a4b5c"
            //   POST /v1/charges → ch_A committed before socket timeout fires
            //   StripeException(SocketTimeoutException) thrown → Promise fails
            //
            // Re-subscription 1 (retry attempt 1):
            //   UUID = "c9d8e7f6-5a4b-3c2d-1e0f-9a8b7c6d5e4f"  ← different UUID
            //   POST /v1/charges → Stripe sees new idempotency key → creates ch_B ← duplicate
            //
            // Re-subscription 2 (retry attempt 2):
            //   UUID = "b2a1c3d4-e5f6-7a8b-9c0d-1e2f3a4b5c6d"  ← yet another UUID
            //   POST /v1/charges → Stripe creates ch_C ← triplicate
            String idempotencyKey = UUID.randomUUID().toString();

            RequestOptions options = RequestOptions.builder()
                .setIdempotencyKey(idempotencyKey)
                .build();
            ChargeCreateParams params = ChargeCreateParams.builder()
                .setAmount(amountCents)
                .setCurrency("usd")
                .setCustomer(customerId)
                .build();

            return Charge.create(params, options);
        })
        .retry(3, Duration.ofSeconds(1));
        // retry(3, Duration.ofSeconds(1)) re-subscribes Blocking.get() on each failure.
        // Each re-subscription re-dispatches the factory lambda to the blocking thread pool.
        // Each factory invocation calls UUID.randomUUID() — fresh UUID per invocation.
    }
}

// Execution timeline for customer "cust_123", billingPeriod="2026-11":
// 10:00:00.000  Promise subscribed — Blocking.get() factory dispatched to blocking pool
// 10:00:00.001  Blocking thread: UUID_A generated; POST /v1/charges sent
// 10:00:29.999  Stripe: ch_A committed (charges.created event fired)
// 10:00:30.001  Socket read timeout fires — StripeException thrown from factory
// 10:00:30.001  Promise.retry() catches failure — waits Duration.ofSeconds(1)
// 10:00:31.001  Re-subscription: Blocking.get() factory dispatched again
// 10:00:31.002  Blocking thread: UUID_B generated — new idempotency key
// 10:00:31.100  POST /v1/charges — Stripe sees new key — creates ch_B
// Result: customer "cust_123" billed twice for November 2026.

The distinction between Promise-construction time and Promise-subscription time is central to understanding why the failure occurs. When a developer writes Blocking.get(() -> { String key = UUID.randomUUID()... }), the UUID.randomUUID() call is syntactically inside the lambda body — it is not evaluated when Blocking.get() is called, but when the lambda is invoked. Blocking.get() stores a reference to the lambda (the Factory<T> functional interface implementation), and invokes it each time the resulting Promise is subscribed. Promise.retry() re-subscribes the resulting Promise. Therefore UUID.randomUUID() is called once per subscription, including retries.

Why constructing the Promise outside the retry does not help if UUID.randomUUID() is inside the factory lambda

A common attempted fix is to construct the Blocking.get() Promise before the retry call, believing that the factory was already “evaluated” at construction time:

// STILL UNSAFE: the factory lambda body is not evaluated at construction time.
// Calling Blocking.get(factory) returns a Promise that stores the factory reference.
// Promise.retry() re-subscribes the stored Promise — re-invokes the factory.
// UUID.randomUUID() inside the factory body evaluates on each invocation.

// The promise object is constructed once, but its factory is called on each subscription.
Promise<Charge> p = Blocking.get(() -> {
    String key = UUID.randomUUID().toString(); // still inside the factory lambda
    return Charge.create(buildParams(customerId, amountCents),
                         RequestOptions.builder().setIdempotencyKey(key).build());
});

// Wrapping in a local variable and then calling .retry() on it changes nothing.
// p is a description of the computation; retry re-runs the description.
return p.retry(3, Duration.ofSeconds(1)); // still re-invokes the factory on each retry

The fix is to compute the idempotency key before constructing the Promise, capture it as an effectively-final local variable, and reference the captured value from inside the factory lambda. The factory lambda will then read the same pre-computed key on every invocation — the captured variable value does not change between Promise subscriptions:

// SAFE: stable content-hash key computed BEFORE constructing the Blocking.get() Promise.
// The key is captured as an effectively-final local variable in the enclosing scope.
// The factory lambda closes over the variable reference — reads the same value
// on every invocation regardless of how many times Promise.retry() re-subscribes.

import org.apache.commons.codec.digest.DigestUtils;
import ratpack.exec.Blocking;
import ratpack.exec.Promise;

import java.time.Duration;

public class BillingHandler {

    public Promise<Charge> chargeCustomer(String customerId, long amountCents,
                                           String billingPeriod) {

        // SAFE: stable key computed here — before Blocking.get() is called.
        // This is evaluated once when chargeCustomer() is invoked, not per subscription.
        // sha256(customerId:billingPeriod:ratpack-billing)[:32] is a deterministic function
        // of stable billing fields — same value regardless of retry attempt number,
        // blocking thread identity, or Promise subscription timestamp.
        final String stableKey = DigestUtils.sha256Hex(
            customerId + ":" + billingPeriod + ":ratpack-billing"
        ).substring(0, 32);

        return Blocking.get(() -> {
            // SAFE: stableKey is a captured effectively-final variable.
            // The factory lambda closes over stableKey — reads the same String reference
            // on each invocation. Promise.retry() re-subscribes and re-invokes this
            // factory, but stableKey is not re-evaluated — it was computed once above.
            //
            // Re-subscription 1: stableKey = "a3f1b2c4d5e6f7a8b9c0d1e2f3a4b5c6" → ch_A (cached)
            // Re-subscription 2: stableKey = "a3f1b2c4d5e6f7a8b9c0d1e2f3a4b5c6" → ch_A (cached)
            // Re-subscription 3: stableKey = "a3f1b2c4d5e6f7a8b9c0d1e2f3a4b5c6" → ch_A (cached)
            // Stripe idempotency cache returns ch_A on every attempt with the same key.
            RequestOptions options = RequestOptions.builder()
                .setIdempotencyKey(stableKey)
                .build();
            ChargeCreateParams params = ChargeCreateParams.builder()
                .setAmount(amountCents)
                .setCurrency("usd")
                .setCustomer(customerId)
                .build();
            return Charge.create(params, options);
        })
        .retry(3, Duration.ofSeconds(1));
    }
}

// Key construction rule: sha256(customerId:billingPeriod:ratpack-billing)[:32]
// — a deterministic function of the two stable billing fields.
// Same output regardless of:
//   - Which Blocking.get() subscription is currently executing
//   - Which retry attempt number (1, 2, or 3)
//   - Which thread in the blocking thread pool is running
//   - What time the current subscription started
//   - Which pod in the Kubernetes cluster is executing the Promise chain

The rule is: evaluate UUID.randomUUID() (or any non-deterministic key source) exactly once, before constructing any Promise or lambda that will be re-invoked under retry. The re-subscription boundary is the outermost .retry() call in the Promise chain. Everything upstream of that call — including Blocking.get() factory lambdas, flatMap() lambdas, and Promise.async() factory lambdas — is re-executed on each re-subscription.

Failure mode 2: flatMap lambda inside an outer Promise.retry() — UUID.randomUUID() captured outside the inner Blocking.get() but inside the outer flatMap lambda — re-executed on each outer re-subscription — ch_B per customer

A more subtle version of the failure arises from flatMap composition when the retry boundary is placed on an outer Promise that wraps a flatMap call. A developer writing a billing run that processes a list of customers often builds a chain like this:

// UNSAFE: UUID.randomUUID() inside the flatMap lambda body —
// even though it is placed outside the inner Blocking.get() supplier.
//
// The outer Promise returned by getActiveCustomers() is wrapped in a flatMap.
// An outer .retry() is placed on the final Promise, intending to retry the
// entire billing run if the database fetch fails.
//
// The developer places UUID.randomUUID() outside Blocking.get(), believing
// it is computed once per customer (at flatMap evaluation time, not per retry).
// In fact, the entire flatMap lambda body re-executes on each outer re-subscription.

import ratpack.exec.Promise;
import ratpack.exec.Blocking;

import java.time.Duration;
import java.util.List;
import java.util.UUID;

public class BillingRunService {

    public Promise<Void> runMonthlyBilling(String billingPeriod) {
        return getActiveCustomers() // Promise<List<Customer>> — may fail on DB timeout
            .flatMap(customers -> {
                // UNSAFE: this lambda body re-executes on each re-subscription of
                // the outer retry boundary placed below on the final Promise.
                //
                // If getActiveCustomers() fails and the outer .retry() re-subscribes
                // the chain, this lambda is called again for each customer — including
                // customers who may have already been billed in the first attempt.
                //
                // Worse: even if getActiveCustomers() succeeds on the first subscription
                // but the Promise returned by this flatMap fails later (e.g. a customer
                // billing Promise fails after several customers have been charged), the
                // outer retry re-subscribes the ENTIRE chain including the flatMap lambda.
                //
                // UUID placed outside Blocking.get() here means: UUID.randomUUID() is
                // called once per customer per flatMap invocation. When the flatMap
                // lambda body re-executes on outer re-subscription, it is called again
                // for every customer — generating fresh UUIDs for customers already billed.
                List<Promise<Charge>> chargePromises = customers.stream()
                    .map(customer -> {
                        // UNSAFE: inside flatMap lambda — re-evaluated per outer retry
                        String key = UUID.randomUUID().toString(); // ← ch_B source

                        return Blocking.get(() -> {
                            // key is captured here — stable across Blocking.get() retries
                            // BUT: not stable across the outer flatMap re-subscription.
                            // On outer retry, the outer flatMap lambda creates a NEW key.
                            return Charge.create(
                                buildParams(customer, amountCents(customer)),
                                RequestOptions.builder().setIdempotencyKey(key).build()
                            );
                        });
                    })
                    .collect(java.util.stream.Collectors.toList());

                return Promise.value(null); // simplified — illustrates the lambda boundary
            })
            .retry(2, Duration.ofSeconds(5)); // outer retry re-subscribes everything above
    }
}

// Execution timeline for two customers "cust_A" and "cust_B":
//
// Outer subscription 1 (initial attempt):
//   getActiveCustomers() → [cust_A, cust_B]
//   flatMap lambda invoked:
//     cust_A: UUID_A1 generated outside Blocking.get()
//     cust_B: UUID_B1 generated outside Blocking.get()
//     cust_A Blocking.get(): ch_A committed before StripeException fires for cust_A
//     StripeException propagates → outer Promise fails
//
// Outer re-subscription 1 (retry attempt 1):
//   getActiveCustomers() → [cust_A, cust_B]   ← re-fetched (DB roundtrip)
//   flatMap lambda re-invoked:
//     cust_A: UUID_A2 generated ← different from UUID_A1 → ch_B for cust_A ← duplicate
//     cust_B: UUID_B2 generated ← different from UUID_B1 (cust_B may not have been charged yet)
//     cust_A Blocking.get(): new key UUID_A2 → Stripe creates ch_B (not a cache hit)
// Result: cust_A billed twice for the billing period.

The failure is structurally identical to failure mode 1 — UUID.randomUUID() is evaluated inside a lambda that is re-called on retry — but it is harder to see because the UUID computation is outside the inner Blocking.get() lambda and appears to be evaluated “once per customer.” It is evaluated once per customer per outer flatMap invocation. The outer .retry() re-invokes the outer flatMap lambda, which in turn re-evaluates the UUID for each customer.

Fix: stable keys computed from stable fields before the outermost retry boundary

The correct fix is to ensure that the idempotency key for each customer is a deterministic function of that customer’s stable fields — not any value computed inside a lambda that sits inside a .retry() boundary. The key must produce the same value whether it is evaluated inside the first outer flatMap invocation or the fifth:

// SAFE: stable keys derived from stable fields — deterministic regardless of
// which flatMap invocation (i.e. which outer retry attempt) is computing them.
//
// sha256(customerId:billingPeriod:ratpack-billing)[:32] produces the same output
// whether called on outer subscription 1 or outer subscription 3.
// The key does not depend on when it is computed — only on what is being billed.

import org.apache.commons.codec.digest.DigestUtils;
import ratpack.exec.Promise;
import ratpack.exec.Blocking;

import java.time.Duration;
import java.util.List;
import java.util.stream.Collectors;

public class BillingRunService {

    public Promise<Void> runMonthlyBilling(String billingPeriod) {
        return getActiveCustomers()
            .flatMap(customers -> {
                // SAFE: key is derived from stable fields — same value regardless of
                // how many times this flatMap lambda has been invoked by outer retries.
                //
                // sha256(customerId:billingPeriod:ratpack-billing) for cust_123 always
                // produces the same 32-character hex string for the same billing period.
                // First flatMap invocation (outer attempt 1): stableKey = "a3f1..."
                // Second flatMap invocation (outer attempt 2): stableKey = "a3f1..." (same)
                // Stripe idempotency cache hits on the same key → returns ch_A, not ch_B.
                List<Promise<Charge>> chargePromises = customers.stream()
                    .map(customer -> {
                        // SAFE: content-hash key — deterministic from stable billing fields.
                        // The formula includes no non-deterministic element:
                        //   no UUID.randomUUID(), no System.currentTimeMillis(),
                        //   no attempt counter, no hostname, no timestamp.
                        final String stableKey = DigestUtils.sha256Hex(
                            customer.getId() + ":" + billingPeriod + ":ratpack-billing"
                        ).substring(0, 32);

                        return Blocking.get(() -> {
                            // stableKey captured from enclosing scope — same value on every
                            // invocation: initial Blocking.get() subscription, inner retries
                            // (if any), and outer flatMap re-invocations from outer retry.
                            return Charge.create(
                                ChargeCreateParams.builder()
                                    .setAmount(customer.getAmountCents())
                                    .setCurrency("usd")
                                    .setCustomer(customer.getStripeId())
                                    .build(),
                                RequestOptions.builder()
                                    .setIdempotencyKey(stableKey)
                                    .build()
                            );
                        });
                    })
                    .collect(Collectors.toList());

                // Promise.all() collects results from all charge Promises.
                // If any charge fails, Promise.all() fails and outer retry re-subscribes.
                // Because stableKeys are content-hash derived, already-charged customers
                // hit Stripe's idempotency cache and return ch_A — no ch_B created.
                return Promise.all(chargePromises).map(charges -> null);
            })
            .retry(2, Duration.ofSeconds(5));
    }
}

// A further improvement: add a pre-flight PostgreSQL guard INSIDE the Blocking.get()
// factory, before the Stripe API call. This ensures customers already billed
// (from a previous outer subscription that partially succeeded) are skipped
// entirely on outer re-subscription — the Stripe call is not made at all.
// The ON CONFLICT DO NOTHING guard survives across Stripe's 24-hour cache TTL,
// crash-recovery restarts, and manual operator-triggered retries.

Blocking.get(() -> {
    boolean inserted = insertBillingAttempt(
        conn, customer.getId(), billingPeriod, stableKey
    );
    if (!inserted) {
        // Row exists — customer was already billed this period.
        // Return the existing charge record instead of calling Stripe again.
        return fetchExistingCharge(conn, customer.getId(), billingPeriod);
    }
    return Charge.create(buildParams(customer), buildOptions(stableKey));
});

The pre-flight ON CONFLICT DO NOTHING guard is the correct defense for outer flatMap retry scenarios because it addresses the case where an outer re-subscription reaches a customer who was successfully billed during a partial first-attempt run. A content-hash stable key alone protects against duplicate Stripe charges within the 24-hour idempotency cache window, but the pre-flight guard also protects against re-billing after the cache expires, operator-triggered re-runs, and crash-recovery scenarios where the process is restarted days after the partial first run.

Failure mode 3: ExecController.executor().scheduleAtFixedRate() fires on every Ratpack node in a multi-pod Kubernetes Deployment — no built-in cluster-wide coordination — distinct UUID.randomUUID() per pod per customer — ch_A, ch_B, and ch_C per customer per billing period

Ratpack applications frequently use the Service lifecycle interface to start background tasks when the server starts. A common pattern for periodic billing is to call ExecController.executor().scheduleAtFixedRate() inside a Service.onStart() implementation:

// UNSAFE: scheduleAtFixedRate() fires on every Ratpack node in the cluster.
// Ratpack's Service interface is called on every application instance.
// ExecController.executor() returns the ScheduledExecutorService that backs
// Ratpack's non-blocking event loop — a raw Java ScheduledExecutorService.
// Scheduling a Runnable on it fires that Runnable on every pod that starts the server.
// In a Kubernetes Deployment with replicas:3, all three pods call onStart(),
// all three schedule billing, and all three fire at the same time.

import ratpack.server.Service;
import ratpack.server.StartEvent;
import ratpack.exec.Execution;
import ratpack.exec.ExecController;

import java.util.UUID;
import java.util.concurrent.TimeUnit;

public class BillingSchedulerService implements Service {

    private final CustomerRepository customerRepository;
    private final BillingService billingService;

    @Override
    public void onStart(StartEvent event) throws Exception {
        ExecController execController = event.getRegistry().get(ExecController.class);

        // This scheduleAtFixedRate() call executes on EVERY Ratpack node.
        // There is no Ratpack-level coordination primitive that limits execution to
        // one node out of N. All three pods in a replicas:3 Deployment call onStart()
        // independently — all three schedule this Runnable — all three fire at the
        // same initial delay and at the same fixed rate.
        execController.executor().scheduleAtFixedRate(() -> {
            Execution.fork().start(exec -> {
                String billingPeriod = java.time.YearMonth.now().toString();
                billingService.runBillingRun(billingPeriod).then(result -> {
                    // billing complete on this pod
                });
            });
        }, 0, 24, TimeUnit.HOURS);
    }
}

// Inside billingService.runBillingRun() — the failure:
public Promise<Void> runBillingRun(String billingPeriod) {
    return getActiveCustomers().flatMap(customers -> {
        List<Promise<Charge>> chargePromises = customers.stream()
            .map(customer -> {
                // UNSAFE: UUID.randomUUID() evaluated independently on each pod.
                // Pod 1: UUID_A1 for cust_123 → ch_A
                // Pod 2: UUID_B1 for cust_123 → ch_B ← simultaneous duplicate
                // Pod 3: UUID_C1 for cust_123 → ch_C ← simultaneous triplicate
                String key = UUID.randomUUID().toString();
                return Blocking.get(() ->
                    Charge.create(buildParams(customer), buildOptions(key))
                );
            })
            .collect(Collectors.toList());
        return Promise.all(chargePromises).map(ignored -> null);
    });
}

// Execution timeline for customer "cust_123", billingPeriod="2026-11":
// All three pods start their scheduleAtFixedRate() Runnable at approximately the
// same time — all three pods were deployed simultaneously via Kubernetes rolling update.
//
// 00:00:00.000  Pod 1: Execution.fork() → runBillingRun() → UUID_A1 for cust_123
// 00:00:00.003  Pod 2: Execution.fork() → runBillingRun() → UUID_B1 for cust_123
// 00:00:00.007  Pod 3: Execution.fork() → runBillingRun() → UUID_C1 for cust_123
// 00:00:00.900  Stripe: ch_A committed (Pod 1), ch_B (Pod 2), ch_C (Pod 3)
// Result: customer "cust_123" billed three times for November 2026.

An important nuance about stable idempotency keys in multi-pod scenarios: replacing UUID.randomUUID() with a content-hash stable key does not fully solve the multi-pod problem. When three pods compute the same stable key for the same customer and simultaneously send three POST /v1/charges requests with that key to Stripe, Stripe may process all three before the first is committed to the idempotency cache. Stripe’s idempotency guarantee applies to sequential requests: a second request with the same key that arrives after the first is committed returns the cached result. It does not guarantee deduplication of concurrent requests arriving within milliseconds of each other, before any of them has been written to the cache. The correct multi-pod fix is cluster-wide coordination at the application layer, before the Stripe API call is made.

Fix: pg_try_advisory_lock() at the billing run entry point for cluster-wide serialization, combined with pre-flight ON CONFLICT DO NOTHING as a permanent guard

// SAFE: pg_try_advisory_lock() serializes billing across the cluster.
// Pre-flight ON CONFLICT DO NOTHING as permanent guard for crash recovery and re-runs.
// Stable content-hash key passed as final variable to Blocking.get() factory.

import ratpack.exec.Blocking;
import ratpack.exec.Promise;
import ratpack.server.Service;
import ratpack.server.StartEvent;
import ratpack.exec.Execution;
import ratpack.exec.ExecController;

import org.apache.commons.codec.digest.DigestUtils;

import javax.sql.DataSource;
import java.sql.Connection;
import java.sql.PreparedStatement;
import java.sql.ResultSet;
import java.sql.SQLException;
import java.time.Duration;
import java.time.YearMonth;
import java.util.List;
import java.util.concurrent.TimeUnit;
import java.util.stream.Collectors;

public class BillingSchedulerService implements Service {

    private final DataSource dataSource;
    private final CustomerRepository customerRepository;

    // Stable integer advisory lock key for the billing function.
    // All pods sharing the same PostgreSQL instance contend on the same key.
    private static final long BILLING_LOCK_KEY = "ratpack-monthly-billing".hashCode() & 0x7FFFFFFFL;

    @Override
    public void onStart(StartEvent event) throws Exception {
        ExecController execController = event.getRegistry().get(ExecController.class);

        execController.executor().scheduleAtFixedRate(() -> {
            Execution.fork().start(exec -> {
                String billingPeriod = YearMonth.now().toString(); // e.g. "2026-11"
                runBillingRunWithLock(billingPeriod).then(result -> {
                    // Either billing completed (lock acquired) or
                    // another pod held the lock (skipped gracefully).
                });
            });
        }, 0, 24, TimeUnit.HOURS);
    }

    private Promise<Void> runBillingRunWithLock(String billingPeriod) {
        return Blocking.get(() -> {
            // Acquire advisory lock — non-blocking (pg_try_advisory_lock, not pg_advisory_lock).
            // Returns true if this pod acquired the lock.
            // Returns false if another pod already holds it — this pod skips immediately.
            // Lock is released automatically when the JDBC connection is closed.
            try (Connection conn = dataSource.getConnection()) {
                try (PreparedStatement lockStmt = conn.prepareStatement(
                        "SELECT pg_try_advisory_lock(?)")) {
                    lockStmt.setLong(1, BILLING_LOCK_KEY);
                    ResultSet rs = lockStmt.executeQuery();
                    rs.next();
                    boolean lockAcquired = rs.getBoolean(1);

                    if (!lockAcquired) {
                        // Another pod is running billing — skip gracefully.
                        System.out.println("Billing lock held by another pod — skipping.");
                        return (Void) null;
                    }

                    try {
                        runBillingUnderLock(conn, billingPeriod);
                    } finally {
                        // Explicit unlock before connection returns to pool.
                        try (PreparedStatement unlockStmt = conn.prepareStatement(
                                "SELECT pg_advisory_unlock(?)")) {
                            unlockStmt.setLong(1, BILLING_LOCK_KEY);
                            unlockStmt.executeQuery();
                        }
                    }
                }
            }
            return (Void) null;
        });
    }

    private void runBillingUnderLock(Connection conn, String billingPeriod)
            throws Exception {

        // Lock acquired — only this pod reaches here.
        // All other pods returned above when pg_try_advisory_lock() returned false.
        List<Customer> customers = customerRepository.findAllActiveSync(conn);

        for (Customer customer : customers) {
            // SAFE: content-hash stable key computed from stable billing fields.
            // Same value on this pod and all other pods for the same customer
            // and billing period — regardless of which pod reaches this point.
            final String stableKey = DigestUtils.sha256Hex(
                customer.getId() + ":" + billingPeriod + ":ratpack-billing"
            ).substring(0, 32);

            // Pre-flight guard: INSERT with ON CONFLICT DO NOTHING.
            // Returns 1 (inserted, this customer has not been billed this period)
            // or 0 (row exists, this customer was already billed — skip).
            // This guard fires even if the advisory lock was not acquired,
            // and survives crash recovery and manual re-runs beyond Stripe's 24h cache.
            boolean inserted = insertBillingAttempt(
                conn, customer.getId(), billingPeriod, stableKey
            );
            if (!inserted) {
                continue; // already billed this period — skip
            }

            try {
                // Synchronous Stripe call inside runBillingUnderLock()
                // which is already inside Blocking.get() — safe to call blocking SDKs.
                // stableKey is a final local — stable across any crash recovery
                // or re-run that reaches this point.
                RequestOptions options = RequestOptions.builder()
                    .setIdempotencyKey(stableKey)
                    .build();
                ChargeCreateParams params = ChargeCreateParams.builder()
                    .setAmount(customer.getAmountCents())
                    .setCurrency("usd")
                    .setCustomer(customer.getStripeId())
                    .build();
                Charge.create(params, options);

            } catch (Exception e) {
                // Mark billing attempt as failed — does not delete the pre-flight row.
                // The pre-flight row ensures this customer is not re-attempted
                // in this same billing run or the next retry of the scheduling task.
                markBillingFailed(conn, customer.getId(), billingPeriod);
                System.err.println("Billing failed for customer " + customer.getId()
                    + ": " + e.getMessage());
            }
        }
    }

    private boolean insertBillingAttempt(Connection conn, String customerId,
                                          String billingPeriod, String idempotencyKey)
            throws SQLException {
        try (PreparedStatement stmt = conn.prepareStatement(
                "INSERT INTO billing_attempts (customer_id, billing_period, idempotency_key, created_at) " +
                "VALUES (?, ?, ?, NOW()) " +
                "ON CONFLICT (customer_id, billing_period) DO NOTHING")) {
            stmt.setString(1, customerId);
            stmt.setString(2, billingPeriod);
            stmt.setString(3, idempotencyKey);
            return stmt.executeUpdate() == 1;
        }
    }

    private void markBillingFailed(Connection conn, String customerId, String billingPeriod)
            throws SQLException {
        try (PreparedStatement stmt = conn.prepareStatement(
                "UPDATE billing_attempts SET status = 'failed' " +
                "WHERE customer_id = ? AND billing_period = ?")) {
            stmt.setString(1, customerId);
            stmt.setString(2, billingPeriod);
            stmt.executeUpdate();
        }
    }
}

// Schema for billing_attempts table:
// CREATE TABLE billing_attempts (
//   customer_id     TEXT        NOT NULL,
//   billing_period  TEXT        NOT NULL,
//   idempotency_key TEXT        NOT NULL,
//   created_at      TIMESTAMPTZ NOT NULL,
//   status          TEXT        NOT NULL DEFAULT 'pending',
//   CONSTRAINT billing_attempts_pkey PRIMARY KEY (customer_id, billing_period)
// );
// UNIQUE constraint on (customer_id, billing_period) makes ON CONFLICT DO NOTHING
// fire on duplicate pairs regardless of idempotency_key value.

// Kubernetes Deployment recommendation: single-replica billing worker.
// The most operationally simple fix is to deploy the BillingSchedulerService
// in a separate Deployment with replicas:1. The advisory lock is then optional.
//
// apiVersion: apps/v1
// kind: Deployment
// metadata:
//   name: ratpack-billing-scheduler
// spec:
//   replicas: 1    # single instance — scheduleAtFixedRate fires once per interval
//   # The ON CONFLICT DO NOTHING guard remains as defense against crash recovery.
//
// apiVersion: apps/v1
// kind: Deployment
// metadata:
//   name: ratpack-api
// spec:
//   replicas: 3    # high availability for API serving — no BillingSchedulerService here

Ratpack’s ExecController.executor() is the compute executor that backs the event loop, not a cluster-aware scheduler. Scheduling a Runnable on it has the same per-JVM semantics as scheduling on any ScheduledExecutorService: the task fires on the current JVM instance, isolated from all other JVM instances. For background tasks in a multi-pod Kubernetes deployment, the advisory lock at the database layer is the correct cluster-wide coordination mechanism, since the database is the only component shared across all pods.

Why Stripe’s 24-hour idempotency cache is not the last line of defense

Stripe’s idempotency cache matches requests by key within a 24-hour sliding window. A request with key K submitted at 10:00:00 UTC will cache-hit for subsequent requests with the same key K submitted before 10:00:00 UTC the following day. After 24 hours, Stripe treats a new request with the same key as a fresh charge.

For monthly billing, crash-recovery scenarios where the application restarts more than 24 hours after the original billing attempt bypass Stripe’s idempotency cache entirely. The stable content-hash key matches nothing in the expired cache, and Stripe creates a new charge. The pre-flight ON CONFLICT DO NOTHING guard is immune to this: it checks your own database, which retains billing records permanently. The advisory lock is also immune: it prevents the billing function from reaching the Stripe API call at all when billing is already in progress or completed for the period.

There is also a nuance specific to Ratpack’s async execution model: Ratpack Promises execute on non-blocking threads, and Blocking.get() dispatches to a separate blocking thread pool. If the process crashes after the Blocking.get() supplier has dispatched to a blocking thread (and the Stripe API call has been sent) but before the Promise resolves and the caller persists the result, the billing record may not be in the database. The pre-flight INSERT ... ON CONFLICT DO NOTHING uses a write-before-call ordering: the billing record is inserted into the database before the Stripe API call is made. If the process crashes after the insert but before the Stripe call, the pre-flight row prevents a re-run from attempting the charge. The row can be cleaned up manually or by a recovery job if the Stripe charge was never created.

Guard Prevents TTL Scope
pg_try_advisory_lock() Concurrent billing runs (multi-pod scheduleAtFixedRate race) Until connection closes or explicit unlock Cluster-wide (same PostgreSQL instance)
ON CONFLICT DO NOTHING pre-flight Duplicate billing attempts from all causes including crash recovery Permanent (row retained in table) Cluster-wide (same PostgreSQL instance)
Stable content-hash key captured before retry boundary Duplicate Stripe charges from Promise.retry() re-subscriptions 24 hours (Stripe cache) Stripe’s API layer
Vault key spend cap at expected_total × 1.10 Financial damage from any combination of the above failures Per billing period (configurable) Proxy layer (Keybrake)

Vault keys for Ratpack billing services

A scoped vault key issued per billing period through a spend-cap proxy provides a hard financial ceiling regardless of the application-layer guards above. The vault key is configured with a daily USD cap set at expected_total_charges × 1.10 — 10% headroom above the expected billing run total. If any combination of the three failure modes above results in charges beyond the expected total, the proxy rejects additional POST /v1/charges requests at the vendor layer. The cap fires independently of the Ratpack application logic, the Stripe idempotency cache, and the PostgreSQL advisory lock state. It is the backstop that fires in exactly the failure path that was not tested before deployment.

// Vault key configuration for a Ratpack billing service.
// Create one vault key per billing period (monthly or daily, depending on billing cadence).
// Set the daily_usd_cap at expected_charges_total * 1.10.
// Set allowed_endpoints to POST /v1/charges and GET /v1/charges/:id only —
// allowlist prevents the Stripe key from being used for any other Stripe endpoint.
// expires_at matches the end of the billing period.

// Keybrake vault key policy for November 2026 billing run:
{
  "vault_key": "vault_key_xxx",
  "vendor": "stripe",
  "daily_usd_cap": 11000.00,       // expected total: $10,000 — cap at 110%
  "allowed_endpoints": [
    "POST /v1/charges",
    "GET /v1/charges/:id"
  ],
  "expires_at": "2026-11-30T23:59:59Z"
}

// Ratpack billing service uses the vault key as the bearer token.
// The proxy at proxy.keybrake.com/stripe/v1/charges forwards to Stripe,
// enforces the daily_usd_cap, logs every request to the audit table,
// and hard-stops any request that would breach the cap.
// If Promise.retry() fires three times and creates ch_A, ch_B, and ch_C,
// the vault cap fires when the cumulative total of ch_A + ch_B + ch_C
// exceeds daily_usd_cap — limiting financial damage even when all
// application-layer guards fail simultaneously.

// BillingService uses vault_key_xxx as the Stripe API key:
RequestOptions options = RequestOptions.builder()
    .setIdempotencyKey(stableKey)
    .setApiKey("vault_key_xxx")       // vault key issued for this billing period
    .setBaseUrl("https://proxy.keybrake.com/stripe/")  // route through spend-cap proxy
    .build();

Implementation checklist for Ratpack and Stripe billing

Put a spend cap on your Ratpack billing service

Keybrake issues scoped vault keys for Stripe, Twilio, and Resend — with per-vendor daily spend caps, allowlisted endpoints, and a one-click kill switch. A Ratpack service using Promise.retry() on a Blocking.get() chain or scheduleAtFixedRate() on a three-replica Kubernetes deployment gets a hard financial ceiling even when the idempotency logic has a bug in a retry path that was never exercised in staging.