gRPC-Kotlin Coroutine Client and Stripe Integration: Restricted API Keys, Spend Caps, and Agent Governance

The Java gRPC post covered three failure modes that arise from gRPC’s ClientInterceptor API, bidirectional StreamObserver reconnects, and the built-in hedging policy. gRPC-Kotlin generates a completely different stub API: every unary RPC becomes a suspend fun, every server-streaming RPC returns Flow<Response>, and every client- or bidirectional-streaming RPC accepts Flow<Request> as input. This coroutine-native model introduces three Stripe billing failure modes that do not appear in the Java version: (1) suspend fun retry re-entry — a manual retry loop or a kotlinx.coroutines retry wrapper re-invokes the billing suspend fun per attempt, and UUID.randomUUID() at function entry produces UUID_B when the first attempt’s charge committed before StatusException(UNAVAILABLE) was thrown; (2) CoroutineContext.Element-based idempotency key — developers store the idempotency key in a custom CoroutineContext.Element and read it from coroutineContext inside the interceptor, but each retry wraps the attempt in withContext(StripeKeyElement(UUID.randomUUID())), constructing a new element with UUID_B; (3) Flow-based bidirectional stream retry — the .retry() operator on the response Flow re-collects the request Flow passed to the stub, re-executing any UUID.randomUUID() in the request producer and generating UUID_B per reconnect. This post covers all three with gRPC-Kotlin 1.x and gRPC 1.6x code, content-hash idempotency keys, CoroutineContext propagation, pre-flight database guards, and vault key spend caps as the financial backstop.

Why gRPC-Kotlin coroutine stubs are structurally different from Java gRPC stubs

The Java gRPC stub for a unary billing RPC looks like this: stub.chargeCustomer(request, responseObserver) — the call is asynchronous via StreamObserver, and retry is typically implemented by wrapping the stub call in an AbstractRetryingStub or by building a retry loop that calls stub.chargeCustomer() again on onError(). The gRPC-Kotlin stub for the same RPC is: suspend fun chargeCustomer(request: ChargeRequest): ChargeResponse — a direct, synchronous-looking suspend fun call that suspends the current coroutine until the response arrives or a StatusException is thrown.

This difference matters for idempotency in three ways. First, the suspension point is the entire RPC call — the suspend fun suspends from the moment the request is sent until the response arrives; a StatusException thrown from it means the underlying channel received an error response or the connection dropped. Whether Stripe committed a charge before that error depends on where in the HTTP/2 lifecycle the error occurred, not on any property of the coroutine. Second, because retry is typically a loop or a wrapper that calls the same suspend fun again, any value evaluated inside the function body is re-evaluated per invocation — including UUID.randomUUID(). Third, the Flow-based streaming API reifies the request source as a Kotlin Flow; the .retry() operator on the response side triggers re-collection of the request Flow, not just reconnection of a channel, which means any value generated lazily inside the request flow is re-generated per retry.

Failure mode 1: suspend fun billing function with retry loop — UUID.randomUUID() at function entry — re-entry on retry produces UUID_B — ch_B when ch_A committed before StatusException thrown

The most common Kotlin pattern for a billing service that calls gRPC looks like this: a suspend fun chargeCustomer(customerId: String, amountCents: Int): String function that builds the request (including generating an idempotency key), calls the coroutine stub, and returns the charge ID. Retry is either a while loop or a retryWhen / repeat wrapper around the function call. When UUID.randomUUID() is evaluated at the top of chargeCustomer(), it is re-evaluated on every call — including every retry invocation:

// BillingService.kt — UNSAFE: UUID.randomUUID() at function entry,
// retry loop re-invokes chargeCustomer() per attempt — UUID_B on first retry.
import io.grpc.StatusException
import java.util.UUID

class BillingService(private val stub: BillingServiceGrpcKt.BillingServiceCoroutineStub) {

    suspend fun chargeCustomer(customerId: String, amountCents: Int): String {
        // BUG: UUID evaluated at entry. Every call to chargeCustomer()
        // — including retries — produces a different UUID.
        val idempotencyKey = UUID.randomUUID().toString()

        val request = ChargeRequest.newBuilder()
            .setCustomerId(customerId)
            .setAmountCents(amountCents)
            .setIdempotencyKey(idempotencyKey) // UUID_A on first call, UUID_B on retry
            .build()

        return stub.chargeCustomer(request).chargeId
    }
}

// Retry site: calls chargeCustomer() again on StatusException(UNAVAILABLE).
// Each call re-enters the function body and re-evaluates UUID.randomUUID().
suspend fun chargeWithRetry(
    service: BillingService,
    customerId: String,
    amountCents: Int
): String {
    var lastException: Exception? = null
    repeat(3) { attempt ->
        try {
            return service.chargeCustomer(customerId, amountCents)
            // ^ First attempt: UUID_A sent. If Stripe committed ch_A before
            // the channel returned UNAVAILABLE (e.g., HTTP/2 RST_STREAM
            // received after the full request was transmitted), calling
            // chargeCustomer() again produces UUID_B — Stripe creates ch_B.
        } catch (e: StatusException) {
            if (e.status.code == io.grpc.Status.Code.UNAVAILABLE) {
                lastException = e
                if (attempt < 2) kotlinx.coroutines.delay(500L * (attempt + 1))
            } else throw e
        }
    }
    throw lastException ?: error("unreachable")
}

The critical window is narrow but real: gRPC’s HTTP/2 transport can receive UNAVAILABLE (or translate an HTTP 503 into it) after the full request has been transmitted to and processed by Stripe. Stripe’s documentation explicitly warns that UNAVAILABLE from a network error on the return path may mean the charge was committed. A retry of chargeCustomer(customerId, amountCents) with UUID_B creates ch_B alongside the already-committed ch_A.

Subtler variant: kotlinx.coroutines retryWhen inline billing lambda

The kotlinx.coroutines library does not provide a suspend fun-level retry utility directly, but teams often implement one by wrapping the billing call in a Flow and using flow { emit(chargeCustomer(...)) }.retryWhen { ... }.first(), or by writing a generic suspend fun <T> retryOnUnavailable(block: suspend () -> T): T helper. In the inline lambda form, the same bug surfaces:

// BillingService.kt — UNSAFE: retryOnUnavailable re-invokes the lambda.
// Lambda generates UUID.randomUUID() on each invocation — UUID_B on retry.
suspend fun retryOnUnavailable(maxAttempts: Int, block: suspend () -> String): String {
    var lastEx: StatusException? = null
    repeat(maxAttempts) { attempt ->
        try { return block() }
        catch (e: StatusException) {
            if (e.status.code == io.grpc.Status.Code.UNAVAILABLE) {
                lastEx = e
                if (attempt < maxAttempts - 1) kotlinx.coroutines.delay(200L * (attempt + 1))
            } else throw e
        }
    }
    throw lastEx ?: error("unreachable")
}

// Usage site — UNSAFE:
val chargeId = retryOnUnavailable(3) {
    // This lambda body is executed on every attempt.
    // UUID.randomUUID() is evaluated fresh on every lambda invocation.
    val idempotencyKey = UUID.randomUUID().toString() // UUID_B on retry
    val request = ChargeRequest.newBuilder()
        .setCustomerId(customerId)
        .setAmountCents(amountCents)
        .setIdempotencyKey(idempotencyKey)
        .build()
    stub.chargeCustomer(request).chargeId
}

The developer may read retryOnUnavailable { ... } as “retry the call” and mentally separate the “call” (the stub invocation) from the “setup” (the key generation). But the retry helper re-invokes the entire lambda, including any setup code inside it. Moving UUID.randomUUID() outside the lambda, or using a stable content-hash key, is the only safe option.

Fix: compute the stable idempotency key before the retry boundary — thread it as a parameter or close over it

// BillingService.kt — FIXED.
import java.security.MessageDigest
import java.nio.charset.StandardCharsets

fun stableIdempotencyKey(customerId: String, period: String): String {
    val input = "$customerId:$period:grpc-kotlin-billing"
    val digest = MessageDigest.getInstance("SHA-256").digest(
        input.toByteArray(StandardCharsets.UTF_8)
    )
    return digest.take(16).joinToString("") { "%02x".format(it) }
}

// chargeCustomer now accepts the stable key as a parameter.
// The caller is responsible for computing the key exactly once,
// before any retry loop, and passing the same value on every attempt.
suspend fun chargeCustomer(
    customerId: String,
    amountCents: Int,
    idempotencyKey: String // computed by caller before retry loop
): String {
    val request = ChargeRequest.newBuilder()
        .setCustomerId(customerId)
        .setAmountCents(amountCents)
        .setIdempotencyKey(idempotencyKey) // same value on every attempt
        .build()
    return stub.chargeCustomer(request).chargeId
}

// Retry site — FIXED:
suspend fun chargeWithRetry(
    service: BillingService,
    customerId: String,
    billingPeriod: String,
    amountCents: Int
): String {
    // Key computed once, outside and before the retry loop.
    // stableIdempotencyKey() is a pure function; same inputs → same output.
    val idempotencyKey = stableIdempotencyKey(customerId, billingPeriod)

    var lastException: Exception? = null
    repeat(3) { attempt ->
        try {
            // Same idempotencyKey passed on every attempt.
            // Stripe receives UUID_A on attempts 1, 2, and 3.
            return service.chargeCustomer(customerId, amountCents, idempotencyKey)
        } catch (e: StatusException) {
            if (e.status.code == io.grpc.Status.Code.UNAVAILABLE) {
                lastException = e
                if (attempt < 2) kotlinx.coroutines.delay(500L * (attempt + 1))
            } else throw e
        }
    }
    throw lastException ?: error("unreachable")
}

Two properties matter here. First, stableIdempotencyKey() is a deterministic pure function: it returns the same 32-character hex string for the same (customerId, billingPeriod) pair, on any JVM instance, in any pod, at any time. If two concurrent coroutines both start a billing run for cus_123 in 2026-09, they both compute the same key; Stripe’s idempotency layer serializes the two requests and the second caller receives the cached result from the first. Second, the idempotencyKey value is closed over by the repeat loop, not regenerated inside it. Every invocation of chargeCustomer() within the loop passes the same closed-over string. This property is preserved regardless of how the retry helper is structured, because the key is computed at the callsite of the retry helper, not inside the helper or its lambda.

Failure mode 2: CoroutineContext.Element-based idempotency key — new element constructed per withContext() retry — UUID_B installed in context — interceptor reads UUID_B from coroutine context — ch_B

Kotlin coroutines propagate structured data through the CoroutineContext. A common pattern in gRPC-Kotlin services is to store request-scoped metadata — trace IDs, tenant IDs, and idempotency keys — in a custom CoroutineContext.Element subclass, then read that element from coroutineContext inside the interceptor’s suspend-capable overrides or from a suspend helper that builds CallOptions before calling the stub. This avoids thread-local storage and integrates cleanly with structured concurrency. The failure mode arises when the retry strategy wraps each attempt in withContext(StripeKeyElement(UUID.randomUUID())), constructing a new element with a new UUID per withContext() invocation:

// StripeKeyElement.kt — a CoroutineContext.Element carrying the idempotency key.
data class StripeKeyElement(val idempotencyKey: String) : CoroutineContext.Element {
    companion object Key : CoroutineContext.Key<StripeKeyElement>
    override val key: CoroutineContext.Key<*> get() = Key
}

// StripeKeyInterceptor.kt — reads the key from coroutineContext via gRPC-Kotlin
// CoroutineContextClientInterceptor pattern, injects it into CallOptions Metadata.
class StripeKeyInterceptor : ClientInterceptor {
    override fun <ReqT, RespT> interceptCall(
        method: MethodDescriptor<ReqT, RespT>,
        callOptions: CallOptions,
        next: Channel
    ): ClientCall<ReqT, RespT> {
        // gRPC-Kotlin propagates the current coroutine context into ClientInterceptor
        // via GrpcContextElement; reading StripeKeyElement from that context
        // returns the value installed by the most recent withContext() call.
        val key = callOptions.getOption(STRIPE_KEY_OPTION)
        return next.newCall(method, callOptions).also { call ->
            // attach key to Metadata on start()
        }
    }
}

// BillingService.kt — UNSAFE: withContext() constructs StripeKeyElement(UUID.randomUUID())
// per retry attempt. Each attempt installs a new element with UUID_B.
suspend fun chargeWithRetry(
    stub: BillingServiceGrpcKt.BillingServiceCoroutineStub,
    customerId: String,
    amountCents: Int
): String {
    var lastEx: StatusException? = null
    repeat(3) { attempt ->
        try {
            // BUG: StripeKeyElement constructed with UUID.randomUUID() here.
            // withContext() installs a new coroutine context element per iteration.
            // On retry (attempt 1), UUID.randomUUID() produces UUID_B.
            return withContext(StripeKeyElement(UUID.randomUUID().toString())) {
                val request = ChargeRequest.newBuilder()
                    .setCustomerId(customerId)
                    .setAmountCents(amountCents)
                    .build()
                stub.chargeCustomer(request).chargeId
                // stub reads StripeKeyElement from coroutineContext — UUID_B on attempt 1.
            }
        } catch (e: StatusException) {
            if (e.status.code == io.grpc.Status.Code.UNAVAILABLE) {
                lastEx = e
                if (attempt < 2) kotlinx.coroutines.delay(300L * (attempt + 1))
            } else throw e
        }
    }
    throw lastEx ?: error("unreachable")
}

The structure of this code looks safe at a glance: withContext(StripeKeyElement(...)) appears outside the stub.chargeCustomer() call, so a reader might assume the key is set once. The bug is that StripeKeyElement(UUID.randomUUID()) evaluates the UUID.randomUUID() argument at the point where withContext() is called — which is inside the repeat loop, on every iteration. The withContext() call on attempt 1 constructs a new StripeKeyElement with UUID_B.

Subtler variant: CoroutineScope.async { } retry with fresh scope per launch

A variant arises when retry is implemented by launching a new coroutine via async { } or launch { } rather than by re-entering a withContext() block. If the idempotency key is passed as an element in the CoroutineScope’s context and the retry creates a new scope per attempt, a new key element is constructed per scope:

// BillingRetrier.kt — UNSAFE: CoroutineScope per attempt, new StripeKeyElement per scope.
suspend fun chargeWithScopedRetry(
    stub: BillingServiceGrpcKt.BillingServiceCoroutineStub,
    customerId: String,
    amountCents: Int
): String {
    var lastEx: Exception? = null
    repeat(3) { attempt ->
        try {
            // BUG: StripeKeyElement(UUID.randomUUID()) evaluated per scope construction.
            // Each CoroutineScope created here has a different UUID in its context.
            val result = CoroutineScope(
                Dispatchers.IO + StripeKeyElement(UUID.randomUUID().toString())
            ).async {
                val request = ChargeRequest.newBuilder()
                    .setCustomerId(customerId)
                    .setAmountCents(amountCents)
                    .build()
                stub.chargeCustomer(request).chargeId
            }.await()
            return result
        } catch (e: StatusException) {
            if (e.status.code == io.grpc.Status.Code.UNAVAILABLE) {
                lastEx = e
                if (attempt < 2) kotlinx.coroutines.delay(200L * (attempt + 1))
            } else throw e
        }
    }
    throw lastEx ?: error("unreachable")
}

The developer may have introduced the CoroutineScope-per-attempt pattern to ensure structured concurrency cleanup on failure (cancelling in-flight work on scope cancellation). The correctness intent is sound; only the idempotency key generation placement is wrong. The fix is the same: compute the stable content-hash key before the retry loop and pass it as the element value.

Fix: compute the stable key once, outside the retry boundary — pass it as the CoroutineContext.Element value on every attempt

// BillingService.kt — FIXED: stable key computed once, passed to every withContext().
suspend fun chargeWithRetry(
    stub: BillingServiceGrpcKt.BillingServiceCoroutineStub,
    customerId: String,
    billingPeriod: String,
    amountCents: Int
): String {
    // Key computed once, before the retry loop.
    // stableIdempotencyKey() is a pure function — same value for same inputs.
    val idempotencyKey = stableIdempotencyKey(customerId, billingPeriod)

    var lastEx: StatusException? = null
    repeat(3) { attempt ->
        try {
            // withContext() installs the same StripeKeyElement value on every attempt.
            // The stub reads UUID_A from coroutineContext on attempts 1, 2, and 3.
            return withContext(StripeKeyElement(idempotencyKey)) {
                val request = ChargeRequest.newBuilder()
                    .setCustomerId(customerId)
                    .setAmountCents(amountCents)
                    .build()
                stub.chargeCustomer(request).chargeId
            }
        } catch (e: StatusException) {
            if (e.status.code == io.grpc.Status.Code.UNAVAILABLE) {
                lastEx = e
                if (attempt < 2) kotlinx.coroutines.delay(300L * (attempt + 1))
            } else throw e
        }
    }
    throw lastEx ?: error("unreachable")
}

Two structural improvements in this fix: first, StripeKeyElement(idempotencyKey) constructs the element with the pre-computed stable key — the same string value is passed to StripeKeyElement() on every iteration, so the context element contains the same UUID_A on all attempts; second, because stableIdempotencyKey() is a pure function, a restart of the calling process — a pod crash, a JVM OOM kill, a deployment rollout — will compute the same key again for the same customer and billing period, and Stripe’s idempotency cache will return the already-committed ch_A rather than creating a new charge.

For the CoroutineScope-per-attempt variant, the fix is the same: compute idempotencyKey before the repeat loop and pass it as the element value in every scope construction — CoroutineScope(Dispatchers.IO + StripeKeyElement(idempotencyKey)) — so every scope carries the same context element.

Failure mode 3: Flow-based bidirectional streaming — .retry() operator re-collects request Flow — UUID.randomUUID() in request producer re-evaluated per collection — UUID_B on first reconnect — ch_B for customers already charged before stream failure

gRPC-Kotlin’s bidirectional streaming stub accepts a Flow<BillingRequest> as the request source and returns a Flow<BillingResponse>: suspend fun billingStream(requests: Flow<BillingRequest>): Flow<BillingResponse>. This model is idiomatic for batch billing systems that want to send a stream of charge commands and receive acknowledgements. When the stream fails with StatusException(UNAVAILABLE), the natural coroutine retry is .retry(3) { it is StatusException } on the response flow. The failure mode arises when the request Flow passed to the stub generates idempotency keys lazily inside the flow producer, because .retry() re-collects the entire flow — including the request flow — from scratch:

// BillingStreamService.kt — UNSAFE:
// flow { emit(...UUID.randomUUID()...) } in request producer.
// .retry() re-collects the request flow on reconnect — UUID_B per reconnect.
import kotlinx.coroutines.flow.*
import io.grpc.StatusException

suspend fun runBillingStream(
    stub: BillingServiceGrpcKt.BillingServiceCoroutineStub,
    customers: List<BillingCustomer>
) {
    // BUG: UUID.randomUUID() evaluated inside the flow { } producer block.
    // flow { } is a cold flow: the producer block executes on every collection.
    // .retry() re-collects the flow, re-executing the producer block,
    // re-evaluating UUID.randomUUID() for every customer.
    val requestFlow: Flow<BillingRequest> = flow {
        for (customer in customers) {
            emit(
                BillingRequest.newBuilder()
                    .setCustomerId(customer.id)
                    .setAmountCents(customer.amountCents)
                    .setIdempotencyKey(UUID.randomUUID().toString()) // UUID_B on reconnect
                    .build()
            )
        }
    }

    stub.billingStream(requestFlow)
        .retry(3) { throwable ->
            // Reconnect on UNAVAILABLE. This re-collects requestFlow from the beginning.
            throwable is StatusException &&
                (throwable as StatusException).status.code == io.grpc.Status.Code.UNAVAILABLE
        }
        .collect { response ->
            println("Charged ${response.customerId}: ${response.chargeId}")
        }
    // If the stream fails after processing, say, 500 of 1000 customers
    // on the first collection (UUID_A per customer, 500 charges committed),
    // .retry() re-collects requestFlow from customer[0] with UUID_B per customer.
    // The 500 already-charged customers receive new charge requests with UUID_B
    // — Stripe creates ch_B for each of the 500 already-charged customers.
}

The extent of the damage scales with where in the customer list the stream failure occurred. If the stream processes all 1,000 customers before failing (for example, a StatusException thrown while consuming the last few response acknowledgements), the retry re-sends all 1,000 requests with new UUIDs, risking 1,000 double charges. If the stream fails after 10 customers, the retry risks 10 double charges (for those 10) plus missing charges for the 990 not yet processed on the first collection (those 990 will be charged correctly on the retry, with UUID_B serving as the unique key for their first charge). In either case, the customers processed before the failure are at risk.

Subtler variant: channelFlow { } with a producer coroutine that re-runs on .retry()

Teams building high-throughput billing pipelines sometimes use channelFlow { } instead of flow { } to allow concurrent emissions from multiple coroutines:

// BillingStreamService.kt — UNSAFE: channelFlow { } with UUID inside producer coroutine.
// .retry() re-collects the channelFlow, re-launching the producer coroutine block.
val requestFlow: Flow<BillingRequest> = channelFlow {
    customers.map { customer ->
        launch {
            send(
                BillingRequest.newBuilder()
                    .setCustomerId(customer.id)
                    .setAmountCents(customer.amountCents)
                    .setIdempotencyKey(UUID.randomUUID().toString()) // re-evaluated per launch
                    .build()
            )
        }
    }
}

A developer might reason that channelFlow { } is “hotter” than flow { } and that the producer runs only once. This is incorrect: channelFlow { } is still a cold flow. The producer block starts executing when the flow is first collected, and re-executes when .retry() re-collects it after a failure. The launch { } blocks inside the producer are re-launched on every collection, and each launch call re-evaluates UUID.randomUUID().

A MutableSharedFlow-based request source is different and can be safe if managed correctly: a MutableSharedFlow is a hot flow — it does not replay emissions to new collectors unless a non-zero replay cache is configured. If the MutableSharedFlow is pre-populated with all billing requests before the stub call, and .retry() re-collects the flow (now the SharedFlow), new collectors receive only future emissions — the already-emitted requests are not replayed. This means the retry does not re-send already-processed customers, but it also does not re-send any of the in-flight customers at the time of failure, potentially missing them. The SharedFlow approach eliminates the UUID_B double-charge but introduces a different correctness problem: missed charges for customers whose requests were in the channel buffer when the stream failed. A pre-flight database guard and post-run reconciliation are needed regardless.

Fix: build the request Flow from pre-computed stable keys — generate all keys before the flow is constructed

// BillingStreamService.kt — FIXED:
// Stable keys computed before the flow is built.
// The flow emits pre-built requests; re-collection on .retry() uses the same keys.
import kotlinx.coroutines.flow.*
import io.grpc.StatusException
import java.security.MessageDigest
import java.nio.charset.StandardCharsets

fun stableIdempotencyKey(customerId: String, period: String): String {
    val input = "$customerId:$period:grpc-kotlin-stream-billing"
    val digest = MessageDigest.getInstance("SHA-256").digest(
        input.toByteArray(StandardCharsets.UTF_8)
    )
    return digest.take(16).joinToString("") { "%02x".format(it) }
}

suspend fun runBillingStream(
    stub: BillingServiceGrpcKt.BillingServiceCoroutineStub,
    customers: List<BillingCustomer>,
    billingPeriod: String
) {
    // Pre-compute all stable keys before building the flow.
    // stableIdempotencyKey() is a pure function: same output for same inputs.
    // These strings are immutable and captured by the flow builder by value.
    val preBuiltRequests: List<BillingRequest> = customers.map { customer ->
        BillingRequest.newBuilder()
            .setCustomerId(customer.id)
            .setAmountCents(customer.amountCents)
            .setIdempotencyKey(stableIdempotencyKey(customer.id, billingPeriod))
            // UUID_A for customer cus_001, UUID_B for cus_002, etc.
            // But the same UUID_A for cus_001 on every collection of this flow.
            .build()
    }

    // The request flow emits pre-built immutable request objects.
    // Re-collection on .retry() emits the same objects with the same stable keys.
    val requestFlow: Flow<BillingRequest> = preBuiltRequests.asFlow()

    stub.billingStream(requestFlow)
        .retry(3) { throwable ->
            throwable is StatusException &&
                (throwable as StatusException).status.code == io.grpc.Status.Code.UNAVAILABLE
        }
        .collect { response ->
            println("Charged ${response.customerId}: ${response.chargeId}")
        }
    // If the stream fails after processing 500 customers and .retry() re-collects,
    // the 500 already-charged customers receive requests with the same stable keys.
    // Stripe returns the cached ch_A for each of those 500 customers — no ch_B.
    // The remaining 500 customers are processed on the retry collection with their
    // own stable keys, each creating one charge.
}

Three properties of this fix: (1) stableIdempotencyKey(customerId, billingPeriod) returns the same value for the same customer and billing period on every JVM instance and every retry — if two pods both start the same billing stream concurrently, they produce the same keys for each customer; Stripe serializes the first request for each key and returns the cached result to the second; (2) preBuiltRequests.asFlow() is a cold flow that emits the same pre-built BillingRequest objects on every collection — the BillingRequest protobuf message is an immutable value; emitting the same object on retry emits the same idempotency key; (3) the .retry() operator re-processes all customers from the beginning — for the 500 already charged on the first collection, Stripe’s idempotency cache returns their existing charge IDs; for the 500 not yet processed, the retry creates new charges.

A pre-flight database guard (see below) closes the remaining edge case: if two concurrent billing runs reach Stripe before the idempotency cache settles, the pre-flight INSERT ... ON CONFLICT DO NOTHING ensures exactly one run wins per customer per billing period, and the other is skipped before the request is sent to the stream.

Shared fix layer: pre-flight database guard and vault key spend cap

Stable content-hash idempotency keys are necessary but not sufficient. Two scenarios remain that the key alone cannot close:

Scenario A: two pods start the same billing run simultaneously. Both compute the same stableIdempotencyKey("cus_123", "2026-09") and both attempt to call the gRPC stub. Stripe’s idempotency layer serializes concurrent same-key requests and returns the cached result to the second — no double charge in this scenario, but the race window is real on a distributed billing system. A pre-flight database guard eliminates the race:

-- Pre-flight insert. Exactly one pod wins the insert; others get 0 rows affected.
INSERT INTO billing_runs (customer_id, billing_period, idempotency_key, created_at)
VALUES (?, ?, ?, NOW())
ON CONFLICT (customer_id, billing_period) DO NOTHING;

-- rows_affected == 0: another pod already owns this billing run. Skip.
-- rows_affected == 1: this pod owns it. Proceed to gRPC stub call.
-- After the response arrives, update the row:
UPDATE billing_runs SET grpc_charge_id = ?, completed_at = NOW()
WHERE customer_id = ? AND billing_period = ?;

Scenario B: the billing job crashes mid-stream and a new coroutine instance starts a new run before the first instance’s work has been reconciled. Stable keys prevent double charges for customers whose billing completed successfully before the crash (Stripe returns the cached charge ID). The pre-flight guard prevents re-billing any customer whose billing_runs row was inserted before the crash, even if grpc_charge_id has not yet been written. Rows with grpc_charge_id IS NULL after a crash can be reconciled via a Stripe API lookup by the stored idempotency_key:

-- Reconcile uncommitted billing_runs after a crash:
-- For each row where grpc_charge_id IS NULL and created_at < NOW() - INTERVAL '5 minutes':
-- GET https://api.stripe.com/v1/charges?idempotency_key=<stored_key>
-- If a charge exists: UPDATE billing_runs SET grpc_charge_id = returned_charge_id
-- If no charge exists: the gRPC call never reached Stripe; re-issue.

Vault key spend cap: set the vault key cap at expected_total_charges × 1.10. A 10% buffer absorbs legitimate retries (a transient UNAVAILABLE on a 10,000-customer stream that requires one retry per customer adds at most 10,000 additional requests, each returning the cached result from Stripe — no new charges, but each attempt counts against the daily API rate limit, not against the financial cap). If a bug in the retry logic causes actual new charges beyond the cap (a UUID_B bug that somehow survives the pre-flight guard), the proxy returns 429 before those charges reach Stripe. The cap is a financial backstop, not a substitute for correct idempotency key discipline.

gRPC-Kotlin interceptor placement for idempotency key injection

If your team prefers to inject the idempotency key at the interceptor level rather than in the billing service or in the BillingRequest proto field, the correct placement in the gRPC-Kotlin interceptor chain is as a ClientInterceptor that reads the key from a CoroutineContext.Element pre-computed by the caller. The interceptor’s interceptCall() method fires once per stub call; it cannot use suspend fun directly, but it can read gRPC Context values that were attached before the call:

// StripeIdempotencyInterceptor.kt — reads stable key from gRPC Context,
// not from UUID.randomUUID() — fires once per stub call, not per retry attempt.
import io.grpc.*

val IDEMPOTENCY_KEY_CONTEXT_KEY: Context.Key<String> =
    Context.key("stripe-idempotency-key")

class StripeIdempotencyInterceptor : ClientInterceptor {
    override fun <ReqT, RespT> interceptCall(
        method: MethodDescriptor<ReqT, RespT>,
        callOptions: CallOptions,
        next: Channel
    ): ClientCall<ReqT, RespT> {
        val key = IDEMPOTENCY_KEY_CONTEXT_KEY.get(Context.current())
            ?: error("No idempotency key in gRPC context — caller must set one")

        return object : ForwardingClientCall.SimpleForwardingClientCall<ReqT, RespT>(
            next.newCall(method, callOptions)
        ) {
            override fun start(responseListener: Listener<RespT>, headers: Metadata) {
                headers.put(
                    Metadata.Key.of("Idempotency-Key", Metadata.ASCII_STRING_MARSHALLER),
                    key
                )
                super.start(responseListener, headers)
            }
        }
    }
}

// Caller sets the stable key in gRPC Context before entering the retry loop.
// gRPC-Kotlin propagates the current gRPC Context through coroutine suspensions
// via GrpcContextElement (added to coroutine context automatically when using
// grpc-kotlin-stub in a coroutine).
suspend fun chargeWithInterceptorRetry(
    stub: BillingServiceGrpcKt.BillingServiceCoroutineStub,
    customerId: String,
    billingPeriod: String,
    amountCents: Int
): String {
    val idempotencyKey = stableIdempotencyKey(customerId, billingPeriod)

    // Attach the stable key to the gRPC Context before the retry loop.
    val grpcContext = Context.current().withValue(IDEMPOTENCY_KEY_CONTEXT_KEY, idempotencyKey)

    var lastEx: StatusException? = null
    repeat(3) { attempt ->
        try {
            return grpcContext.call {
                // The StripeIdempotencyInterceptor reads idempotencyKey from Context.
                // Context.current() inside call { } returns grpcContext.
                // interceptCall() fires once per stub call and reads the same key
                // from the same grpcContext on every attempt.
                val request = ChargeRequest.newBuilder()
                    .setCustomerId(customerId)
                    .setAmountCents(amountCents)
                    .build()
                stub.chargeCustomer(request).chargeId
            }
        } catch (e: StatusException) {
            if (e.status.code == io.grpc.Status.Code.UNAVAILABLE) {
                lastEx = e
                if (attempt < 2) kotlinx.coroutines.delay(500L * (attempt + 1))
            } else throw e
        }
    }
    throw lastEx ?: error("unreachable")
}

The critical property: grpcContext is built once, before the retry loop, with the stable idempotencyKey string. grpcContext.call { ... } attaches grpcContext as the current gRPC context for the duration of the lambda. On each retry attempt, Context.current() inside the interceptor returns grpcContext — the same context, with the same stable key. The interceptor reads UUID_A on all three attempts.

Summary table

Further reading

• gRPC and Stripe Integration — Java gRPC ClientInterceptor retry; bidirectional StreamObserver reconnect; hedging policy sends concurrent RPC copies to different pods
• Reactor Netty and Stripe Integration — HttpClient.headers(Consumer<HttpHeaders>) consumer called per request-send; Mono.retryWhen() re-subscribes with new UUID per attempt; Mono.timeout() fires after Stripe committed the charge
• Spring WebFlux and Stripe Integration — WebClient ExchangeFilterFunction retry generates new UUID per filter invocation; Retry.backoff() on WebClient.ResponseSpec re-sends with UUID_B; reactive error signals after committed charge
• Ktor Client and Stripe Integration — Ktor HttpRequestRetry plugin re-evaluates UUID.randomUUID() in modifyRequest { } per retry; coroutine cancellation vs billing idempotency
• Apache HttpClient 5 and Stripe Integration — sync CloseableHttpClient retry via HttpRequestRetryStrategy; FutureRequestExecutionService async patterns with retry