Akka Typed EventSourcedBehavior and Stripe Integration: Restricted API Keys, Spend Caps, and Agent Governance

Akka Typed’s EventSourcedBehavior is used in production billing systems for its event-sourcing guarantees: events persisted to a durable journal, state reconstructed by replaying events on actor recovery, and idempotency enforced by guard conditions in the commandHandler. These guarantees are stronger than classic-actor supervisor strategies, but three billing failure modes survive the transition to Typed behaviors and surface only in production under pod restarts, rolling deploys, and ask-pattern timeouts between actor hierarchies.

This post covers all three failure modes with Scala Akka Typed 2.9.x code, content-hash idempotency keys stable across EventSourcedBehavior recovery cycles, context.ask() retries, and ClusterSharding rebalancing events; pre-flight PostgreSQL checks — and per-billing-period vault keys via a spend-cap proxy as a hard backstop when the application-layer fix is bypassed. For related failure modes arising from RestartSource and Source.tick() in Akka Streams-based billing pipelines, see the Akka Streams and Stripe Integration post.

Failure mode 1: EventSourcedBehavior recovery finds BillingInProgress state and resumes billing with System.currentTimeMillis() at recovery time — not the initiatedAt timestamp stored in the persisted BillingInitiated event — producing a different idempotency key and ch_B

The standard pattern for Stripe billing in an EventSourcedBehavior actor is to persist a BillingInitiated event before issuing the Stripe call, then persist a BillingCompleted event on Stripe’s success response. Persisting BillingInitiated before calling Stripe ensures that a crash between the Stripe call and the BillingCompleted event can be detected on recovery: the actor recovers, replays events, finds BillingInProgress state from the journal, and knows to reissue the Stripe call. The RecoveryCompleted signal handler is the natural place to trigger this resume.

The failure mode arises when the BillingInitiated event records the initiation wall-clock time (initiatedAt: Long), this timestamp is used to build the Stripe idempotency key in the original commandHandler, and the RecoveryCompleted resume path computes a new timestamp at recovery time rather than reading the initiatedAt field from the replayed event stored in actor state:

import akka.actor.typed.{ActorRef, Behavior}
import akka.actor.typed.scaladsl.{ActorContext, Behaviors}
import akka.persistence.typed.PersistenceId
import akka.persistence.typed.scaladsl.{Effect, EventSourcedBehavior}
import akka.persistence.typed.scaladsl.RecoveryCompleted
import org.apache.commons.codec.digest.DigestUtils

object BillingActor {
  sealed trait Command
  final case class BillCustomer(customerId: String, billingPeriod: String,
                                 replyTo: ActorRef[BillingReply]) extends Command
  final case class ResumePendingBilling(customerId: String,
                                        billingPeriod: String) extends Command

  sealed trait Event
  final case class BillingInitiated(customerId: String, billingPeriod: String,
                                     initiatedAt: Long) extends Event
  final case class BillingCompleted(customerId: String, billingPeriod: String,
                                     chargeId: String) extends Event

  sealed trait BillingReply
  final case class BillingAcknowledged(chargeId: String) extends BillingReply
  final case class BillingAlreadyComplete(chargeId: String) extends BillingReply

  final case class State(
    pendingBilling: Option[(String, String, Long)] = None, // (customerId, billingPeriod, initiatedAt)
    completedBillings: Map[(String, String), String] = Map.empty  // (customerId, billingPeriod) -> chargeId
  )

  def apply(customerId: String, billingPeriod: String,
            stripeClient: StripeClient,
            billingRepo: BillingRepo): Behavior[Command] =
    Behaviors.setup { context =>
      EventSourcedBehavior[Command, Event, State](
        persistenceId = PersistenceId.ofUniqueId(s"billing-$customerId-$billingPeriod"),
        emptyState = State(),
        commandHandler = commandHandler(context, stripeClient, billingRepo),
        eventHandler = eventHandler
      ).receiveSignal {
        case (state, RecoveryCompleted) =>
          state.pendingBilling.foreach { case (cId, period, _) =>
            // Recovery found BillingInProgress — resume the Stripe call.
            // BUG: ResumePendingBilling carries only customerId and billingPeriod.
            // The initiatedAt timestamp is in state.pendingBilling but is NOT passed
            // in the self-message — it will be re-generated in the command handler.
            context.self ! ResumePendingBilling(cId, period)
          }
      }
    }

  private def commandHandler(
    context: ActorContext[Command],
    stripeClient: StripeClient,
    billingRepo: BillingRepo
  ): (State, Command) => Effect[Event, State] = { (state, command) =>
    command match {
      case BillCustomer(customerId, billingPeriod, replyTo) =>
        state.completedBillings.get((customerId, billingPeriod)) match {
          case Some(chargeId) =>
            Effect.reply(replyTo)(BillingAlreadyComplete(chargeId))
          case None =>
            val initiatedAt = System.currentTimeMillis()
            Effect
              .persist(BillingInitiated(customerId, billingPeriod, initiatedAt))
              .thenRun { _ =>
                // Key built with initiatedAt from this run: e.g., sha256("cust_123:2026-08:1748700000000")
                val key = DigestUtils.sha256Hex(
                  s"$customerId:$billingPeriod:$initiatedAt"
                ).take(32)
                // Stripe call happens here — actor may crash before BillingCompleted is persisted.
                stripeClient.chargeAsync(customerId, billingPeriod, key).foreach { chargeId =>
                  context.self ! StoreCompletion(customerId, billingPeriod, chargeId, replyTo)
                }
              }
        }

      case ResumePendingBilling(customerId, billingPeriod) =>
        // UNSAFE: generates a NEW timestamp at resume time.
        // initiatedAt from the persisted BillingInitiated event is sitting in state.pendingBilling
        // but this handler ignores it and calls currentTimeMillis() again.
        // On recovery after a 5-minute downtime: initiatedAt = 1748700300000
        // Original BillingInitiated.initiatedAt stored in state: 1748700000000
        // sha256("cust_123:2026-08:1748700300000") != sha256("cust_123:2026-08:1748700000000")
        // Stripe sees the new key as a fresh request — creates ch_B.
        val newTimestamp = System.currentTimeMillis()
        val key = DigestUtils.sha256Hex(
          s"$customerId:$billingPeriod:$newTimestamp"
        ).take(32)
        Effect.none.thenRun { _ =>
          stripeClient.chargeAsync(customerId, billingPeriod, key) // ch_B
        }
    }
  }
}

The crash scenario: the actor receives BillCustomer, persists BillingInitiated(customerId="cust_123", billingPeriod="2026-08", initiatedAt=1748700000000) to the event journal, and issues the Stripe call. Stripe processes ch_A (charge ch_A9f3b) and returns the response. The actor is about to send StoreCompletion to itself to persist BillingCompleted when the JVM is killed by Kubernetes pod eviction. The event journal contains BillingInitiated but not BillingCompleted.

On recovery, the eventHandler replays BillingInitiated and populates state.pendingBilling = Some(("cust_123", "2026-08", 1748700000000)). The RecoveryCompleted handler finds the pending billing and sends ResumePendingBilling("cust_123", "2026-08") to context.self. The ResumePendingBilling handler computes newTimestamp = System.currentTimeMillis() at the time of processing the self-message — a different millisecond than the 1748700000000 stored in the event. sha256("cust_123:2026-08:1748700300000")[:32] does not match sha256("cust_123:2026-08:1748700000000")[:32]. Stripe’s idempotency cache has no record of the new key. ch_B is created. Customer 123 is charged twice for August 2026.

The failure is invisible in unit tests because the recovery path is rarely tested with live Stripe idempotency key validation, and in integration tests where the actor is restarted the initiatedAt delta is typically a few milliseconds — still different, but the test may use mocked Stripe clients that don’t enforce key uniqueness.

The fix for failure mode 1

The idempotency key must be stable across the initial charge attempt and every recovery-triggered resume. The initiatedAt timestamp stored in the BillingInitiated event is available in state.pendingBilling after recovery. The ResumePendingBilling message should carry it, or the command handler should read it from state rather than re-generating it. Better: use a content-hash key derived from stable business fields only, so the key is identical whether computed in the original BillCustomer handler or the recovery ResumePendingBilling handler:

object BillingActor {
  // ResumePendingBilling carries the initiatedAt from the persisted event
  final case class ResumePendingBilling(customerId: String,
                                        billingPeriod: String,
                                        initiatedAt: Long) extends Command

  private def commandHandler(
    context: ActorContext[Command],
    stripeClient: StripeClient,
    billingRepo: BillingRepo
  ): (State, Command) => Effect[Event, State] = { (state, command) =>
    command match {
      case BillCustomer(customerId, billingPeriod, replyTo) =>
        state.completedBillings.get((customerId, billingPeriod)) match {
          case Some(chargeId) =>
            Effect.reply(replyTo)(BillingAlreadyComplete(chargeId))
          case None =>
            val initiatedAt = System.currentTimeMillis()
            // Content-hash key — does NOT depend on initiatedAt.
            // Identical whether computed here or in ResumePendingBilling handler.
            // Must NOT include: initiatedAt (changes per actor restart),
            //   System.nanoTime() (new ns per JVM start),
            //   UUID.randomUUID() inside commandHandler (new UUID per call),
            //   PersistenceId sequence number (changes on re-run after snapshot deletion),
            //   context.self.path.toString (different on shard rebalance to new node).
            val key = DigestUtils.sha256Hex(
              s"$customerId:$billingPeriod:akka-billing"
            ).take(32)
            // Pre-flight: claim the billing slot before calling Stripe.
            // On recovery resume, this returns 0 rows if the original Stripe call succeeded
            // and the billing record was committed. Skips the Stripe call — prevents ch_B.
            val preFlightResult = billingRepo.insertIfAbsent(customerId, billingPeriod, key)
            Effect
              .persist(BillingInitiated(customerId, billingPeriod, initiatedAt))
              .thenRun { _ =>
                if (preFlightResult > 0) {
                  stripeClient.chargeAsync(customerId, billingPeriod, key).foreach { chargeId =>
                    context.self ! StoreCompletion(customerId, billingPeriod, chargeId, replyTo)
                  }
                } else {
                  context.log.info("Billing pre-flight found existing record for {} {} — skipping Stripe",
                    customerId, billingPeriod)
                }
              }
        }

      case ResumePendingBilling(customerId, billingPeriod, _initiatedAt) =>
        // Reads initiatedAt from the persisted event (passed via self-message from RecoveryCompleted)
        // but does NOT use it in the key — content-hash from business fields only.
        // Same key as the original BillCustomer handler regardless of when recovery runs.
        val key = DigestUtils.sha256Hex(
          s"$customerId:$billingPeriod:akka-billing"
        ).take(32)
        // Pre-flight check: if the original Stripe call succeeded (ch_A committed),
        // this returns 0 rows — skip Stripe, persist nothing, recovery is complete.
        val preFlightResult = billingRepo.insertIfAbsent(customerId, billingPeriod, key)
        Effect.none.thenRun { _ =>
          if (preFlightResult == 0) {
            context.log.info("Billing resume: pre-flight found committed record for {} {} — ch_A already exists",
              customerId, billingPeriod)
          } else {
            stripeClient.chargeAsync(customerId, billingPeriod, key).foreach { chargeId =>
              context.self ! StoreCompletion(customerId, billingPeriod, chargeId, null)
            }
          }
        }
    }
  }

  // RecoveryCompleted handler passes initiatedAt from state.pendingBilling
  def apply(...): Behavior[Command] =
    Behaviors.setup { context =>
      EventSourcedBehavior[Command, Event, State](...).receiveSignal {
        case (state, RecoveryCompleted) =>
          state.pendingBilling.foreach { case (cId, period, initiatedAt) =>
            // Pass initiatedAt so the command handler has it — even though the key
            // does not use it, passing it maintains the full event context for logging.
            context.self ! ResumePendingBilling(cId, period, initiatedAt)
          }
      }
    }
}

-- Pre-flight table — same schema as the Akka Streams post
CREATE TABLE billing_records (
    customer_id      TEXT NOT NULL,
    billing_period   TEXT NOT NULL,
    idempotency_key  TEXT NOT NULL,
    status           TEXT NOT NULL DEFAULT 'pending',
    charge_id        TEXT,
    created_at       TIMESTAMPTZ NOT NULL DEFAULT now(),
    CONSTRAINT billing_records_pk PRIMARY KEY (idempotency_key),
    CONSTRAINT billing_records_uq UNIQUE (customer_id, billing_period)
);

INSERT INTO billing_records (customer_id, billing_period, idempotency_key)
VALUES ($1, $2, $3)
ON CONFLICT (customer_id, billing_period) DO NOTHING
RETURNING 1;

The recovery path is now safe. If the original Stripe call succeeded and the billing record was committed before the actor crashed, the pre-flight check on recovery returns 0 rows and skips Stripe. If the original Stripe call never reached Stripe (actor crashed in thenRun before the HTTP request was issued), the pre-flight check returns 1 row (new insert), and the Stripe call is issued with the same content-hash key that would have been used in the original attempt — Stripe creates ch_A fresh, and the billing record is updated with the returned charge ID.

Failure mode 2: billing scheduler actor retries a timed-out context.ask() — each ask allocates a unique ephemeral replyTo: ActorRef[BillingReply] whose path ends up in the billing entity’s idempotency key — the first ask created ch_A; the second ask creates ch_B

Akka Typed’s context.ask() is a request-response pattern that sends a message to a target actor and delivers the response to the calling actor as a typed message. Every context.ask() call allocates a unique ephemeral child actor — the replyTo: ActorRef[BillingReply] passed in the request message — that lives only for the duration of the ask timeout. The ephemeral actor has a unique path: akka://billing-system/user/billing-scheduler/$$a, akka://billing-system/user/billing-scheduler/$$b, and so on, incrementing per ask call. On timeout, the ephemeral actor is stopped, and the response message (if it arrives late) is dead-lettered.

The failure mode occurs when the billing entity uses the replyTo ActorRef’s path, hashCode, or any derived value in the Stripe idempotency key — a pattern that appears in codebases that generate idempotency keys from request envelopes rather than from request body content:

// BillingSchedulerActor: sends BillCustomer to each billing entity via context.ask()
// and retries on timeout.

object BillingSchedulerActor {
  sealed trait Command
  case object TriggerBillingCycle extends Command
  final case class BillingEntityResult(customerId: String, result: Try[BillingEntityReply]) extends Command

  def apply(shardRegion: ActorRef[ShardingEnvelope[BillingEntityCommand]],
            customers: List[String],
            billingPeriod: String): Behavior[Command] =
    Behaviors.receive { (context, message) =>
      message match {
        case TriggerBillingCycle =>
          customers.foreach { customerId =>
            // Each context.ask() allocates a unique replyTo: ActorRef[BillingEntityReply]
            // Path: akka://billing-system/user/billing-scheduler/$$a (first ask)
            //        akka://billing-system/user/billing-scheduler/$$b (second ask on timeout)
            context.ask(
              shardRegion,
              (replyTo: ActorRef[BillingEntityReply]) =>
                ShardingEnvelope(customerId, BillCustomer(customerId, billingPeriod, replyTo))
            ) {
              case Success(reply) =>
                BillingEntityResult(customerId, Success(reply))
              case Failure(timeout: AskTimeoutException) =>
                // Retry on timeout — the entity may have been rebalancing
                // Sends a SECOND BillCustomer with a different replyTo ($$b, $$c, etc.)
                context.ask(
                  shardRegion,
                  (replyTo2: ActorRef[BillingEntityReply]) =>
                    ShardingEnvelope(customerId, BillCustomer(customerId, billingPeriod, replyTo2))
                ) {
                  case Success(reply2) => BillingEntityResult(customerId, Success(reply2))
                  case Failure(_)      => BillingEntityResult(customerId, Failure(new Exception("retry failed")))
                }
                BillingEntityResult(customerId, Failure(timeout))
            }
          }
          Behaviors.same
      }
    }
}

// BillingEntity: UNSAFE key derivation using replyTo ActorRef path
object BillingEntity {
  final case class BillCustomer(customerId: String, billingPeriod: String,
                                 replyTo: ActorRef[BillingEntityReply]) extends BillingEntityCommand

  def commandHandler(stripeClient: StripeClient): (State, BillingEntityCommand) => Effect[Event, State] =
    { (state, command) =>
      command match {
        case BillCustomer(customerId, billingPeriod, replyTo) =>
          // UNSAFE: replyTo.path.toString is unique per context.ask() call.
          // First ask: replyTo.path = "akka://billing-system/user/billing-scheduler/$$a"
          // Second ask (timeout retry): replyTo.path = "akka://billing-system/user/billing-scheduler/$$b"
          // sha256("cust_123:2026-08:akka://billing-system/user/billing-scheduler/$$a:billing")[:32]
          //   != sha256("cust_123:2026-08:akka://billing-system/user/billing-scheduler/$$b:billing")[:32]
          // First ask created ch_A. Second ask creates ch_B.
          val requestCorrelation = replyTo.path.toString
          val key = DigestUtils.sha256Hex(
            s"$customerId:$billingPeriod:$requestCorrelation:billing"
          ).take(32)

          Effect
            .persist(BillingInitiated(customerId, billingPeriod, key))
            .thenRun { _ =>
              stripeClient.chargeAsync(customerId, billingPeriod, key).onComplete {
                case Success(chargeId) =>
                  replyTo ! BillingSucceeded(chargeId)
                case Failure(ex) =>
                  replyTo ! BillingFailed(ex.getMessage)
              }
            }
      }
    }
}

The timeout scenario during a rolling deploy: the billing scheduler sends the first BillCustomer for customer 123 to the sharded billing entity at T=0. At T=200ms, the billing entity begins processing: it calls the Stripe API (ch_A) and is about to call replyTo ! BillingSucceeded. At T=205ms, the Kubernetes rolling deploy triggers a shard rebalancing — the billing entity’s shard is migrating from node A to node B. Akka Sharding sends a graceful Passivate message to the entity. The entity is mid-Stripe-call; it defers the passivation until the handler completes. The BillingSucceeded reply is sent to the ephemeral replyTo actor at T=210ms. The scheduler’s ask timeout fires at T=300ms (if the timeout is set to 300ms — a common value for internal actor communication). The scheduler never received the reply because the reply arrived at the ephemeral actor at T=210ms but the scheduler’s timeout handler also fired at T=300ms and marked the ask as failed before the reply was processed (a race between the reply delivery and the timeout callback in the scheduler’s message queue).

The scheduler retries with a second context.ask(). The second ask allocates a new ephemeral replyTo with path .../$$b. The billing entity (now reactivated on node B after the shard migration completes) receives the second BillCustomer. It checks its state via EventSourcedBehavior recovery, finds that the first BillingInitiated event is in the journal but no BillingCompleted event follows it — the entity crashed (was passivated) between persisting BillingInitiated and persisting BillingCompleted. The entity processes the second BillCustomer command through its commandHandler. The command handler builds the key using the new replyTo.path.toString (different from the first ask’s replyTo). Stripe sees the new key, finds no prior charge, and creates ch_B.

The same failure mode occurs when using replyTo.hashCode() (different per JVM allocation), System.identityHashCode(replyTo) (same as hashCode for actors), or any UUID.randomUUID() generated at the BillCustomer message construction site in the scheduler (rather than at the billing entity). Any per-request envelope value that is regenerated on retry creates a different key.

The fix for failure mode 2

The billing entity’s idempotency key must not include any value from the replyTo ActorRef or any per-ask metadata. The key belongs to the billing business event — customer 123, August 2026 — not to the transport mechanism that delivered the billing command. The entity also needs a guard condition in its commandHandler to detect and reject duplicate BillCustomer commands that arrive while billing is already in progress (the state after a BillingInitiated event is replayed but before BillingCompleted is persisted):

// BillingEntity: SAFE key derivation — no replyTo, no per-ask metadata

object BillingEntity {
  def commandHandler(stripeClient: StripeClient,
                     billingRepo: BillingRepo): (State, BillingEntityCommand) => Effect[Event, State] =
    { (state, command) =>
      command match {
        case BillCustomer(customerId, billingPeriod, replyTo) =>
          // Guard 1: already completed billing for this period
          state.completedBillings.get((customerId, billingPeriod)) match {
            case Some(chargeId) =>
              return Effect.reply(replyTo)(BillingAlreadyComplete(chargeId))
            case None =>
          }

          // Guard 2: billing already in progress (BillingInitiated persisted, BillingCompleted not yet)
          // Reject the duplicate BillCustomer command — do not re-issue the Stripe call.
          // The RecoveryCompleted handler (Failure mode 1) handles the resume separately.
          if (state.billingInProgress.contains((customerId, billingPeriod))) {
            context.log.warn("Duplicate BillCustomer received while billing in progress for {} {} — rejecting",
              customerId, billingPeriod)
            return Effect.reply(replyTo)(BillingInProgress())
          }

          // Content-hash key — stable across all retries, all asks, all shard migrations.
          // Must NOT include: replyTo.path.toString (unique per context.ask() call),
          //   replyTo.hashCode() (different JVM allocation per ask),
          //   UUID.randomUUID() at command construction site in the scheduler,
          //   System.currentTimeMillis() at command receipt (different ms per delivery),
          //   context.self.path.toString (changes on shard migration to new node+path),
          //   shard region sender path or routing metadata.
          val key = DigestUtils.sha256Hex(
            s"$customerId:$billingPeriod:akka-billing"
          ).take(32)

          // Pre-flight DB check before persisting and calling Stripe
          val preFlightResult = billingRepo.insertIfAbsent(customerId, billingPeriod, key)

          Effect
            .persist(BillingInitiated(customerId, billingPeriod))
            .thenRun { _ =>
              if (preFlightResult == 0) {
                // Another path already committed billing for this customer+period
                replyTo ! BillingAlreadyComplete("reconciliation-pending")
              } else {
                stripeClient.chargeAsync(customerId, billingPeriod, key).onComplete {
                  case Success(chargeId) =>
                    billingRepo.markCompleted(customerId, billingPeriod, chargeId)
                    context.self ! StoreBillingCompleted(customerId, billingPeriod, chargeId, replyTo)
                  case Failure(ex) =>
                    replyTo ! BillingFailed(ex.getMessage)
                }
              }
            }
      }
    }
}

The billingInProgress guard condition in the commandHandler is the critical addition. When the entity recovers from a state where BillingInitiated is in the journal but BillingCompleted is not, the eventHandler populates state.billingInProgress. Any subsequent BillCustomer command (from a scheduler retry, a manual trigger, or a duplicate delivery) is rejected immediately without calling Stripe. The RecoveryCompleted signal handler is responsible for resuming the in-progress billing exactly once, through the controlled resume path described in failure mode 1. The scheduler that receives BillingInProgress() knows to wait and poll rather than retry immediately.

If the scheduler must retry on timeout (for cases where the entity was truly unreachable during rebalancing and never processed the first ask), include a stable per-request ID in the BillCustomer message that is generated by the scheduler once per customer per billing cycle — not per ask call — and stored persistently in the scheduler’s own event log so it survives scheduler restarts:

// Scheduler generates a stable request ID per customer per billing cycle — not per ask call.
// Stored in the scheduler's own EventSourcedBehavior journal.
final case class BillCustomer(
  customerId: String,
  billingPeriod: String,
  schedulerRequestId: String, // sha256(schedulerId + customerId + billingPeriod)[:16] — stable per cycle
  replyTo: ActorRef[BillingEntityReply]
) extends BillingEntityCommand

// Billing entity uses schedulerRequestId — NOT replyTo — as part of the key context
// (but still ignores it in favor of the content-hash for robustness)
val key = DigestUtils.sha256Hex(
  s"$customerId:$billingPeriod:akka-billing"
).take(32)
// schedulerRequestId is used only for distributed tracing correlation, not in the Stripe key.

Failure mode 3: ClusterSharding billing entity uses Behaviors.withTimers periodic billing tick — shard rebalancing during a rolling deploy creates two entity instances that fire the billing timer in the same billing period — entity activation timestamp in the key produces ch_B on the reactivated entity

A billing entity that manages its own scheduling uses Behaviors.withTimers to register a periodic billing tick at activation time. In a ClusterSharding deployment, entity activation happens whenever a message is first routed to the entity’s shard region on a given node. The timer starts at that moment. A rolling deploy of the Kubernetes deployment that hosts the Akka cluster nodes causes shard rebalancing: shard regions migrate from terminating pods to newly started pods. During rebalancing, an entity that was active on node A is passivated (stopped) and then reactivated on node B. The timer restarts on node B with a new registration time, derived from the wall clock at entity activation on node B.

If the idempotency key includes the entity’s activation timestamp — captured in Behaviors.setup at activation time — the key produced by the entity on node B (reactivated, say, 18 hours after the original entity on node A fired the billing timer and created ch_A) is different. When the new timer fires 30 days after node B’s activation time, it falls within a 30-day window that overlaps with the original billing period. But the key is different because the activation timestamp is different. Stripe creates ch_B:

import akka.actor.typed.scaladsl.{ActorContext, Behaviors, TimerScheduler}
import akka.cluster.sharding.typed.scaladsl.{ClusterSharding, EntityTypeKey, ShardedDaemonProcess}
import scala.concurrent.duration._

object BillingTimerEntity {
  sealed trait Command
  case object BillingTick extends Command
  final case class GetState(replyTo: ActorRef[State]) extends Command

  val TypeKey: EntityTypeKey[Command] = EntityTypeKey[Command]("BillingTimerEntity")

  def apply(customerId: String, stripeClient: StripeClient): Behavior[Command] =
    Behaviors.setup { context =>
      // UNSAFE: activation time captured at entity setup on this node.
      // If entity is reactivated on a new node after shard migration, this is a new timestamp.
      // activationTime on node A: 1748700000000 (entity activated at T=0)
      // activationTime on node B: 1748764800000 (entity activated 18 hours later, after migration)
      val activationTime = System.currentTimeMillis()

      Behaviors.withTimers { timers =>
        // Timer starts from NOW on this node — 30 days from activationTime
        timers.startTimerWithFixedDelay("billing-tick", BillingTick, 30.days)

        Behaviors.receiveMessage {
          case BillingTick =>
            // UNSAFE: activationTime is different on node A vs. node B.
            // Node A fires at T=30d: sha256("cust_123:2026-08:1748700000000:billing")
            // Node B fires at T=18h+30d: timer fired at a wall-clock time 26 hours after node A's fire
            //   — but the billingPeriod "2026-09" is the SAME (August ended, September started)
            //   — Stripe's 24-hour idempotency cache for the August key has expired
            //   — sha256("cust_123:2026-08:1748764800000:billing") is a new key
            //   — Stripe creates ch_B for August 2026
            val key = DigestUtils.sha256Hex(
              s"$customerId:2026-08:$activationTime:billing"
            ).take(32)
            stripeClient.chargeAsync(customerId, "2026-08", key)
            Behaviors.same
        }
      }
    }
}

The shard migration scenario that produces double charges: the billing entity for customer 123 is first activated on node A at T=0 (say, June 1, 2026 at 00:00 UTC). Its timer fires at T+30d (July 1, 2026 at 00:00 UTC) for the June billing period. It creates ch_A with key sha256("cust_123:2026-06:1748700000000:billing")[:32]. At T+18h (June 1, 2026 at 18:00 UTC), a rolling deploy begins. The entity on node A is passivated gracefully. It is reactivated on node B at T+18h with a new activationTime = 1748764800000. The new timer fires at T+18h+30d (July 1, 2026 at 18:00 UTC). The billing period “2026-06” is computed from the current calendar month at fire time — still June. The key is sha256("cust_123:2026-06:1748764800000:billing")[:32]. Stripe’s 24-hour idempotency cache for the July 1 at 00:00 UTC fire has expired by July 1 at 18:00 UTC (18 hours later). Stripe creates ch_B for June 2026.

The failure is subtle because the entity is using a 30-day billing timer and the timer’s absolute fire time changes with each shard migration. An entity migrated once every few months will eventually fire its timer at a wall-clock time far enough from the original fire that Stripe’s idempotency cache provides no protection. The activationTime in the key makes every migration event a potential duplicate charge.

A more immediate variant of this failure occurs during a fast shard migration race: the old entity on node A and the new entity on node B are both briefly active during the handoff window if Akka Sharding uses aggressive rebalancing settings or if the node A entity is slow to passivate (processing a pending Stripe call). Both entities have the same BillingTick timer registered. Both fire within seconds of each other (node A’s remaining interval vs. node B’s new 30-day interval from activation). Both process the billing tick. With an activationTime in the key, both keys are different. Two concurrent Stripe charges within Stripe’s idempotency window appear as two distinct requests. ch_A and ch_B.

The fix for failure mode 3

The idempotency key must not include the entity’s activation timestamp or any per-entity-instance value that changes across shard migrations. Use a content-hash from stable business fields. Additionally, use the EventSourcedBehavior’s event journal as a guard: persist a BillingTickIssued(billingPeriod) event before calling Stripe, and check in the commandHandler whether this billing period has already been issued:

object BillingTimerEntity {
  def apply(customerId: String, stripeClient: StripeClient,
            billingRepo: BillingRepo): Behavior[Command] =
    Behaviors.setup { context =>
      // No per-activation state captured here that would affect the idempotency key.
      Behaviors.withTimers { timers =>
        timers.startTimerWithFixedDelay("billing-tick", BillingTick, 30.days)

        EventSourcedBehavior[Command, Event, State](
          persistenceId = PersistenceId.ofUniqueId(s"billing-timer-$customerId"),
          emptyState = State(),
          commandHandler = { (state, command) =>
            command match {
              case BillingTick =>
                val billingPeriod = currentBillingPeriod() // e.g., "2026-06" from calendar

                // Guard: already issued billing tick for this period in this entity's journal.
                // Survives shard migrations: the journal is persistent and is replayed
                // on the new node after migration. The new entity on node B sees
                // BillingTickIssued("2026-06") in its state and skips the Stripe call.
                if (state.issuedPeriods.contains(billingPeriod)) {
                  context.log.info("BillingTick for period {} already issued — skipping", billingPeriod)
                  return Effect.none
                }

                // Content-hash key — no activation time, no node ID, no entity path.
                // Same key whether computed on node A or node B, regardless of when the entity
                // was activated or migrated.
                val key = DigestUtils.sha256Hex(
                  s"$customerId:$billingPeriod:akka-billing"
                ).take(32)

                // Pre-flight DB check — handles concurrent execution during the migration
                // handoff window where both old and new entity fire BillingTick simultaneously.
                val preFlightResult = billingRepo.insertIfAbsent(customerId, billingPeriod, key)

                Effect.persist(BillingTickIssued(customerId, billingPeriod)).thenRun { _ =>
                  if (preFlightResult == 0) {
                    context.log.info("Pre-flight found existing billing record for {} {} — skipping Stripe",
                      customerId, billingPeriod)
                  } else {
                    stripeClient.chargeAsync(customerId, billingPeriod, key).onComplete {
                      case Success(chargeId) =>
                        billingRepo.markCompleted(customerId, billingPeriod, chargeId)
                        context.self ! StoreBillingCompleted(customerId, billingPeriod, chargeId)
                      case Failure(ex) =>
                        context.log.error("Stripe charge failed for {} {}: {}", customerId, billingPeriod, ex.getMessage)
                    }
                  }
                }
            }
          },
          eventHandler = { (state, event) =>
            event match {
              case BillingTickIssued(_, billingPeriod) =>
                state.copy(issuedPeriods = state.issuedPeriods + billingPeriod)
              case BillingCompleted(_, billingPeriod, chargeId) =>
                state.copy(completedBillings = state.completedBillings + (billingPeriod -> chargeId))
            }
          }
        )
      }
    }

  private def currentBillingPeriod(): String = {
    val now = java.time.YearMonth.now(java.time.ZoneOffset.UTC)
    now.toString // e.g., "2026-06"
  }
}

The issuedPeriods guard closes the timer double-fire during shard migration. The entity on node B, after recovering its journal, finds BillingTickIssued("2026-06") in its replayed state. When the timer fires on node B (at T+18h+30d, which the timer thinks is 30 days after its registration), the commandHandler finds the period already in state.issuedPeriods and returns Effect.none without calling Stripe. The customer is billed exactly once per period regardless of how many times the entity migrates across shard regions.

The pre-flight database check handles the concurrent execution window during fast shard migrations where both old and new entity process BillingTick for the same period simultaneously — before either entity has persisted BillingTickIssued and the issuedPeriods guard is populated. PostgreSQL’s UNIQUE (customer_id, billing_period) constraint serializes the concurrent INSERT ... ON CONFLICT DO NOTHING from both entities: only one succeeds; the other returns 0 rows affected and skips Stripe. The content-hash key ensures that even if both entities somehow reach Stripe before the pre-flight check (a race within the database’s transaction isolation window), Stripe’s idempotency cache deduplicates within 24 hours.

Vault key governance: the per-billing-period spend cap

All three failure modes are addressed at the application layer by content-hash idempotency keys, pre-flight database checks, and event-sourced guard conditions. A vault key per billing period provides a hard backstop when the application-layer fix fails to prevent all duplicate paths: a recovery resume that bypasses the guard condition due to a bug in the State serializer (deserialized issuedPeriods missing entries after schema evolution), a pre-flight check that returns the wrong row count due to a PostgreSQL connection pool misconfiguration, or a fast shard migration race that produces concurrent pre-flight inserts within a single database transaction isolation window.

A vault key is issued to the billing system before the billing cycle begins. The proxy enforces a spend cap of expected_monthly_revenue × 1.10: 10% above the expected billing total for the period. Any Stripe charge that would push the period total above the cap is rejected at the proxy layer before reaching Stripe, with an audit log entry recording the attempt, the customer ID, and the vault key that made the call.

// Issue a vault key for this billing period before any BillingTick actors start.
// In an EventSourcedBehavior billing system, issue the vault key in the billing scheduler actor
// at the start of each billing cycle (persist a VaultKeyIssued event to the scheduler journal).

POST https://proxy.keybrake.com/keys
{
  "vendor": "stripe",
  "billing_period": "2026-08",
  "period_usd_cap": 54890,    // expectedTotal × 1.10 = 49900 × 1.10
  "allowed_endpoints": ["/v1/charges", "/v1/payment_intents"],
  "expires_at": "2026-09-01T00:00:00Z"
}

// Response
{
  "vault_key": "vk_live_xxxxxxxxxx",
  "cap_usd": 54890,
  "vendor": "stripe"
}

// Pass vault_key to each BillingTimerEntity as a configuration value.
// Billing entity uses vault_key as the Authorization bearer in all Stripe calls.
// BillingTimerEntity receives vault_key via ClusterSharding message or configuration actor.

// Duplicate charge attempt from a reactivated entity on node B that bypassed issuedPeriods guard:
// HTTP 429 Too Many Requests — "spend cap exceeded for billing_period 2026-08"
// Proxy audit log: customer_id=cust_123, amount_cents=4990, vault_key=vk_live_xxxxxxxxxx, ts=..., blocked=true

The vault key spend cap is particularly effective against the shard migration failure mode because migration events are operationally visible but not always correlated with billing audit logs. An operations team reviewing Stripe’s charge history sees two charges for the same customer in the same period; the spend cap violation log identifies the vault key and timestamp, which correlates to the shard migration event in the Akka cluster event log. The cap provides both prevention and forensic evidence.

For billing systems with high variance in monthly revenue — enterprise customers with variable usage billing — set the cap at the 99th-percentile monthly revenue rather than expectedTotal × 1.10. A cap set too tight blocks legitimate high-value charges when actual revenue exceeds the expected total. A cap set too loose provides less protection against duplicate charge scenarios. For fixed-price subscription billing (the most common case for SaaS), expected_total × 1.10 is a reasonable backstop: it passes through any single month where a handful of late-added customers push the total above the estimate, while blocking the doubling-of-all-charges scenario that every failure mode in this post produces.

Gap analysis

Snapshot retention deletes BillingCompleted events — recovery starts from snapshot showing billing in-progress

EventSourcedBehavior with withRetention(SnapshotCountRetentionCriteria(keepNSnapshots = 2, snapshotEvery = 100)) snapshots state every 100 events and deletes old events after the snapshot is saved. If the actor snapshots after event 100 (BillingInitiated at sequence 99, BillingCompleted at sequence 100) and then deletes events 1–98, the snapshot at sequence 100 includes both events and recovery correctly shows billing as completed. Safe in this case.

The failure occurs when the actor snapshots after event 99 (BillingInitiated at sequence 99) and before event 100 (BillingCompleted at sequence 100 — which hasn’t happened yet because the actor crashed between the Stripe call and persisting BillingCompleted). The snapshot at sequence 99 shows State(billingInProgress = Set("cust_123:2026-08"), ...). Old events are deleted. On recovery, the actor loads the snapshot (billing in progress), finds no subsequent BillingCompleted event in the truncated journal (because it was never persisted), and enters the resume path. The pre-flight check on resume correctly handles this: if ch_A was created and the billing record is in the database, the pre-flight returns 0 rows and skips Stripe. If the snapshot was taken after BillingInitiated but before the Stripe call was issued (the actor committed the event but hadn’t executed thenRun yet at snapshot time — a race during heavy write load), the billing record is not in the database and the resume correctly issues the Stripe call fresh. The pre-flight check is the authoritative guard; the actor state is the trigger for recovery.

Effect.noReply() leaves callers to retry — duplicate delivery to the billing entity

A billing entity that uses Effect.noReply() (or Effect.none without a reply) leaves the calling scheduler actor with no confirmation that the billing command was processed. The scheduler’s ask times out and it retries. The billing entity receives the second BillCustomer command. If the entity’s commandHandler does not check the billingInProgress and completedBillings guard conditions, it processes the second command and calls Stripe again — with the same content-hash key (safe within 24 hours) or with a different per-delivery key (not safe). Always include a reply to the ask pattern. If the billing entity cannot reply synchronously (the Stripe call is asynchronous), use a deferred reply pattern: save the replyTo in the actor’s state (not the event journal — replyTo ActorRefs are not serializable across JVMs), and reply when StoreCompletion arrives. On recovery where replyTo is lost (not serializable), the scheduler’s ask will time out — but the billingInProgress guard prevents the retry from creating ch_B.

Stash overflow during recovery drops BillCustomer commands

During EventSourcedBehavior recovery, commands are stashed in the behavior’s internal stash buffer while events are being replayed from the journal. The default stash capacity is bounded (typically 1,000 messages). If the billing scheduler sends BillCustomer commands for all 5,000 customers before the billing entity finishes replaying 10,000 historical events from the journal, the stash overflows. Overflow messages are dropped (the default behavior with StashOverflowStrategy.Fail causes the entity to fail; with a custom strategy, messages are silently dropped). Dropped BillCustomer commands mean those customers are never billed for the period — under-charging, not double-charging. But the scheduler, seeing no reply (the ask timed out because the entity was busy replaying), retries. If the entity finishes recovery and is now in completed state for some customers (billed before the crash) and pending state for others, the retried commands may be processed correctly — or may race against ResumePendingBilling self-messages from the RecoveryCompleted handler.

For large billing entity journals, increase the stash capacity via akka.persistence.typed.stash-capacity = 10000 in the Akka configuration, or throttle the billing scheduler to not send all BillCustomer commands at once — use a pull-based pattern where the entity sends a ReadyForNext message to the scheduler after processing each command.

Schema evolution breaks State deserialization — recovery loads snapshot with empty issuedPeriods

The issuedPeriods: Set[String] field added to State in the failure mode 3 fix must be handled in the snapshot serializer when deploying to a cluster that has existing snapshots using the old State schema (without issuedPeriods). If the serializer deserializes an old snapshot and leaves issuedPeriods empty (the None case in a JSON deserializer with no default), the entity’s guard condition shows no issued periods even though billing for the current period was completed months ago. The timer fires, the guard condition passes (empty set), and Stripe is called with the content-hash key. Within Stripe’s 24-hour window, Stripe returns the cached charge. Outside the window (almost certainly, for billing periods from months ago), Stripe creates a new charge. The pre-flight database check is the last line of defense: the billing record is in the database with status = 'completed', so the pre-flight returns 0 rows and Stripe is not called. The pre-flight check must be tested under schema-evolution conditions; it is the backstop when event-sourced guard conditions fail due to deserialization drift.

Summary

All three failure modes share the root cause: idempotency key material generated from actor-lifecycle metadata — recovery timestamps, ephemeral ActorRef paths, entity activation instants — rather than from stable business identity. The Akka Typed event-sourcing model provides strong durability guarantees for events and state, but those guarantees do not automatically extend to the idempotency of external API calls made from thenRun callbacks and timer handlers. The content-hash key derived from customerId + billingPeriod + service-namespace is the correct primitive: it is computed identically in the original command handler, the recovery resume path, the retry path, and the shard-migrated entity’s timer handler. The pre-flight INSERT ... ON CONFLICT DO NOTHING and the event-sourced guard conditions (billingInProgress, issuedPeriods) close the gaps for concurrent execution paths that bypass Stripe’s 24-hour idempotency cache. The vault key spend cap is the operational backstop when all three application-layer guards are bypassed by edge cases in schema evolution, serializer drift, or advisory lock races during high-concurrency billing cycles.