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

Akka Streams is a Reactive Streams implementation used in high-throughput Scala and Java billing backends — stream graphs reading billing triggers from Alpakka Kafka sources, RestartSource wrappers providing resilience against transient failures, and Source.tick() driving periodic billing loops. Its materialization model, supervision semantics, and at-least-once Kafka delivery create three billing failure modes that are invisible in single-node local testing and surface only in production under crash recovery, parallel pod deployments, or stream restarts.

This post covers all three failure modes with Scala Akka Streams 2.9.x code, content-hash idempotency keys stable across RestartSource re-materializations and Alpakka Kafka replays, an Akka Cluster Singleton pattern for Source.tick() billing serialization across pods, 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 by concurrent pod execution within Stripe’s transaction isolation window.

Failure mode 1: RestartSource.withBackoff() re-materializes the source factory on each restart — a run UUID generated inside the factory lambda produces a different idempotency key on restart, billing already-charged customers a second time

Akka Streams’ RestartSource.withBackoff() provides automatic stream restart with exponential backoff. It accepts a factory function — () => Source[T, _] — that is called each time the inner source needs to be (re-)materialized: at initial startup and after every stream failure or completion. The factory is called fresh on each restart with no shared state between invocations.

A billing system that processes all due customers in a single stream often generates a “billing run ID” at stream initialization to correlate log lines and audit records for a single billing cycle. The natural place to generate this run ID is inside the RestartSource factory lambda — the same place where the stream source is constructed. UUID.randomUUID() called inside the factory produces a new UUID on every restart. System.currentTimeMillis() called inside the factory produces a new millisecond timestamp on every restart. If either value feeds into the Stripe idempotency key, the restart run carries different keys for every customer and creates ch_B for all customers billed in the first run before the stream failed:

import akka.NotUsed
import akka.stream.scaladsl._
import akka.stream.RestartSettings
import scala.concurrent.duration._
import java.util.UUID

// UNSAFE: UUID.randomUUID() inside the RestartSource factory lambda
// A new UUID is generated on every stream restart — different idempotency key for each run.
val billingSource: Source[String, NotUsed] = RestartSource.withBackoff(
  RestartSettings(minBackoff = 1.second, maxBackoff = 30.seconds, randomFactor = 0.2)
) { () =>
  // Factory called fresh on each restart.
  // runId = new UUID on startup AND on every restart.
  val runId = UUID.randomUUID().toString

  Source
    .future(customerRepo.findAllDueForBilling())
    .mapConcat(identity)
    .mapAsync(1) { customer =>
      // sha256("cust_123:2026-08:a3f7d9b1-...") on run 1 → ch_A
      // sha256("cust_123:2026-08:b8c2e4f0-...") on restart run 2 → ch_B
      // Customer was already charged in run 1 but the key is different on restart
      val key = DigestUtils.sha256Hex(
        s"${customer.id}:${customer.billingPeriod}:$runId"
      ).take(32)
      stripeClient.chargeCustomer(customer, key)
    }
}

The failure sequence is: run 1 starts, processes customers 1 through 47 (creating charges ch_A₁ through ch_A₄₇), and then the stream fails at customer 48 — a transient database timeout, a brief Stripe API unavailability, or an unhandled exception in the mapAsync handler. RestartSource calls the factory again after the configured backoff. The factory generates a new runId. The new source re-fetches all due customers from the database, because customerRepo.findAllDueForBilling() returns all customers whose billing record has not yet been committed as “billed” in the application database. If the application-layer billing records were not committed before the stream failed (a common pattern: commit after the Stripe call returns successfully), all 500 customers still appear as “due” in the database. The restart run processes all 500 — including customers 1 through 47 who are already charged — with new keys. Stripe has no record of the new keys; it creates ch_B₁ through ch_B₄₇.

The failure is also triggered by system-local values that change between restarts: System.nanoTime() captured at stream start, the ActorMaterializer’s internal flow ID (accessed via Attributes on the materialized value), a counter incremented in the factory lambda that starts at 0 on each JVM start (a restart within the same JVM increments the counter; a JVM restart resets it to 0, giving the same counter value as a previous JVM run for customers 0–N). None of these values are visible in unit tests where the factory is called exactly once and the stream runs to completion without failure.

The fix for failure mode 1

Compute the idempotency key from stable business fields only: the customer ID and the billing period. Do not include any value generated at stream materialization time, any per-run identifier, or any per-ActorSystem state. The key must be identical whether it is computed in run 1, restart run 2, or restart run 50:

import akka.NotUsed
import akka.stream.scaladsl._
import akka.stream.RestartSettings
import scala.concurrent.duration._

val billingSource: Source[String, NotUsed] = RestartSource.withBackoff(
  RestartSettings(minBackoff = 1.second, maxBackoff = 30.seconds, randomFactor = 0.2)
) { () =>
  // No UUID, no currentTimeMillis, no run-local state inside the factory.
  // The source factory is stateless — it produces the same stream shape on every restart.
  Source
    .future(customerRepo.findAllDueForBilling())
    .mapConcat(identity)
    .mapAsync(1) { customer =>
      // Stable key — same on run 1, restart run 2, and restart run 50.
      // Must NOT include: UUID.randomUUID() (new per factory call),
      // System.currentTimeMillis() (new ms per factory call),
      // System.nanoTime() (new ns per factory call),
      // materializer.executionContext.hashCode() (different per ActorSystem instance),
      // any value captured from the outer RestartSource factory scope that changes.
      val key = DigestUtils.sha256Hex(
        s"${customer.id}:${customer.billingPeriod}:akka-billing"
      ).take(32)

      // Pre-flight: claim the billing slot before calling Stripe.
      // If run 1 already billed this customer (and committed the billing record),
      // findAllDueForBilling() will not return them on restart.
      // If run 1 billed but did not commit (crash between Stripe call and DB write),
      // the pre-flight insert returns 0 rows — billing is skipped until reconciliation.
      billingRepo.insertIfAbsent(customer.id, customer.billingPeriod, key).flatMap {
        case 0 => Future.successful("already_billed")
        case _ => stripeClient.chargeCustomer(customer, key)
      }
    }
}

-- billing_records schema
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)
);

-- Pre-flight insert
INSERT INTO billing_records (customer_id, billing_period, idempotency_key)
VALUES ($1, $2, $3)
ON CONFLICT (customer_id, billing_period) DO NOTHING;

The pre-flight check closes the restart failure regardless of what caused the stream to fail. If run 1 billed customers 1–47 and committed billing records for each before failing, findAllDueForBilling() excludes customers 1–47 on the restart run because the billing records exist and mark them as billed. If run 1 crashed between the Stripe call and the billing record commit for customer 47, the pre-flight check on the restart run finds no row for customer 47, inserts one, and re-issues stripe.charges.create() with the same content-hash key — Stripe returns the cached ch_A₄₇ response within the 24-hour window, and the reconciliation job uses the returned charge ID to update the billing record.

Failure mode 2: Alpakka Kafka committableSource with mapAsync(N) — offset batch committed after all N parallel Stripe calls complete; ActorSystem crash between the last Stripe call and the commit replays the entire uncommitted batch, per-run metadata in the key produces ch_B for already-charged records

Alpakka Kafka’s Consumer.committableSource provides at-least-once message delivery. The Kafka offset for a message is committed only after downstream processing — including the Stripe call — completes successfully. This guarantees that a crashed consumer will replay and re-process any message whose offset was not committed, preventing missed billing. The cost is exactly-once delivery at the application layer: messages can be delivered and processed more than once if the consumer crashes after processing but before committing the offset.

The failure mode compounds when mapAsync(parallelism) processes N billing records concurrently and offsets are committed in batches for efficiency. The typical Alpakka Kafka pattern batches committable offsets from multiple records and commits them together:

import akka.kafka.scaladsl.Consumer
import akka.kafka.{CommitterSettings, ConsumerSettings, Subscriptions}
import akka.kafka.scaladsl.Committer
import akka.stream.scaladsl._
import org.apache.kafka.clients.consumer.ConsumerConfig
import org.apache.kafka.common.serialization.StringDeserializer

val parallelism = 4

// UNSAFE: per-ActorSystem startup UUID in the idempotency key
// Different UUID on each JVM start (including crash recovery restart)
val actorSystemStartId = UUID.randomUUID().toString // computed once at JVM start

Consumer
  .committableSource(consumerSettings, Subscriptions.topics("billing-triggers"))
  .mapAsync(parallelism) { message =>
    val record   = message.record
    val customer = Json.parse(record.value()).as[Customer]

    // actorSystemStartId is different after crash recovery restart —
    // JVM restart produces new UUID, different key for same Kafka message on replay.
    // sha256("cust_123:2026-08:f3a9-...:kafka-billing") on first delivery
    // sha256("cust_123:2026-08:a7b1-...:kafka-billing") on replay after crash → ch_B
    val key = DigestUtils.sha256Hex(
      s"${customer.id}:${customer.billingPeriod}:$actorSystemStartId:kafka-billing"
    ).take(32)

    stripeClient
      .chargeCustomer(customer, key)
      .map(_ => message.committableOffset)
  }
  // Batch all 4 offsets — committed together after all 4 Stripe calls return
  .batch(parallelism, first => CommittableOffsetBatch.empty.updated(first))(_.updated(_))
  .mapAsync(1)(_.commitInternal())
  .runWith(Sink.ignore)

The crash scenario: 4 billing records are being processed concurrently by mapAsync(4). Stripe processes ch_A (record 1), ch_B (record 2), and ch_C (record 3) successfully — their futures complete and their committable offsets flow into the batch accumulator. Record 4’s Stripe call is still in flight. At this moment the ActorSystem crashes — an OOM kill, a SIGKILL from Kubernetes pod eviction, or a JVM crash from an unrelated bug. The offset batch for records 1–3 was never committed (it was waiting for record 4 to complete the batch). Kafka’s consumer group coordinator times out the dead consumer after session.timeout.ms and reassigns the partitions. The restarted consumer replays records 1–4 from the last committed offset — which is before record 1.

On replay, actorSystemStartId is a new UUID (different JVM start). The idempotency key for records 1, 2, and 3 is different from the key used in the first delivery. Stripe has no record of the new keys. Three duplicate charges: ch_A′, ch_B′, ch_C′.

The same failure mode occurs with System.currentTimeMillis() captured at consumer start (different ms on restart), with the Kafka consumer group generation ID embedded in the key (generation increments on each rebalance — stable within a session but changes on restart), and with any per-connection or per-session identifier from the Kafka consumer client (connection ID, member ID assigned by the group coordinator — these are opaque strings that change on reconnect).

A more subtle variant: using record.record().offset() in the idempotency key. The Kafka offset is stable — the same message has the same offset on every replay. This looks safe. The failure surfaces when the Kafka topic is compacted: the log compaction process removes older messages with the same key, and the surviving message may have a different offset than the one processed in the original run. In practice, billing trigger topics are usually not compacted (they use append-only semantics), but the assumption is fragile.

The fix for failure mode 2

The idempotency key must be derived from the message’s business payload — the customer ID and billing period extracted from the Kafka record value — not from any metadata that changes between the original delivery and a crash-recovery replay. The Kafka offset, partition, timestamp, and consumer group metadata are all replay-stable for uncompacted topics, but none of them are semantically meaningful for billing idempotency. Compute the key from the business fields:

import akka.kafka.scaladsl.Consumer
import akka.kafka.scaladsl.Committer
import akka.kafka.{CommitterSettings, ConsumerSettings, Subscriptions}
import akka.stream.scaladsl._

val parallelism = 4

Consumer
  .committableSource(consumerSettings, Subscriptions.topics("billing-triggers"))
  .mapAsync(parallelism) { message =>
    val record   = message.record
    val customer = Json.parse(record.value()).as[Customer]

    // Stable key — same on first delivery and on every crash-recovery replay.
    // Must NOT include: actorSystemStartId (new UUID per JVM start),
    // System.currentTimeMillis() at consumer start or at record processing time (different ms),
    // record.record().timestamp() formatted as a string (stable for non-compacted topics,
    //   but semantically wrong: the timestamp is the producer's wall clock, not the billing period),
    // consumer group generation ID (increments on rebalance — changes on restart),
    // Kafka consumer member ID (assigned by group coordinator — changes on reconnect),
    // any per-connection UUID from the Kafka client library's internal state.
    val key = DigestUtils.sha256Hex(
      s"${customer.id}:${customer.billingPeriod}:akka-billing"
    ).take(32)

    // Pre-flight: claim the billing slot before calling Stripe.
    // If this record was already processed in a previous delivery (crash recovery),
    // the pre-flight insert returns 0 rows — skip Stripe, commit the offset.
    billingRepo.insertIfAbsent(customer.id, customer.billingPeriod, key).flatMap {
      case 0 =>
        // Already billed — commit offset without calling Stripe again
        Future.successful(message.committableOffset)
      case _ =>
        stripeClient
          .chargeCustomer(customer, key)
          .map(_ => message.committableOffset)
    }
  }
  // Commit per-record for billing — do not batch across customers.
  // Batching delays the commit window, increasing replay scope on crash.
  // For billing, per-record commit overhead is acceptable.
  .mapAsync(1)(_.commitInternal())
  .runWith(Sink.ignore)

Per-record commit (rather than batched commit) is a deliberate trade-off for billing streams. Batching reduces Kafka commit overhead by amortizing round-trips across N records, which matters for high-throughput consumer groups processing millions of events per second. For billing streams — which process hundreds or thousands of customer charges per billing cycle, not millions per second — the per-record commit overhead is acceptable. The benefit is a shorter replay window: a crash at customer 47 replays from offset 47, not from offset 44 (the last batch boundary). With the pre-flight check, even a replay of the full uncommitted batch is safe; the per-record commit is belt-and-suspenders against a pre-flight edge case under PostgreSQL’s transaction isolation.

If per-record commit is unacceptable for throughput reasons, use a batch commit with a short batch timeout: batch(maxBatch = 20, timeout = 100.millis). This limits the replay scope to at most 20 records or 100ms of messages, whichever comes first, rather than waiting for all N concurrent mapAsync futures to complete before committing any offset.

Failure mode 3: Source.tick() fires independently on every Kubernetes pod — no cluster-wide coordination, billing loop runs N times simultaneously in an N-pod deployment

Source.tick(initialDelay, interval, tick) creates a source that emits one element per interval, driven by the ActorSystem’s scheduler. Like Vert.x’s vertx.setPeriodic(), it is per-ActorSystem with no distributed coordination. In a Kubernetes Deployment with replicas: 3, three ActorSystem instances each run an independent Source.tick() timer. All three fire at (approximately) the same wall-clock time — within pod startup jitter and NTP drift, typically within a few seconds of each other. Each pod runs the complete billing loop for all customers:

import akka.actor.ActorSystem
import akka.stream.scaladsl._
import scala.concurrent.duration._

implicit val system: ActorSystem = ActorSystem("billing-system")

// UNSAFE: Source.tick() in a multi-pod Kubernetes Deployment
// Fires on THIS pod's ActorSystem — no coordination with other pods.
// With replicas: 3, three concurrent billing loops run for every customer.
Source
  .tick(initialDelay = 1.minute, interval = 30.days, tick = "billing-trigger")
  .mapAsync(1) { _ =>
    customerRepo.findAllDueForBilling()
  }
  .mapConcat(identity)
  .mapAsync(4) { customer =>
    // sys.env.getOrElse("HOSTNAME", "unknown") = pod name from Kubernetes downward API:
    //   billing-pod-7f8b9c-abc on pod A
    //   billing-pod-7f8b9c-def on pod B
    //   billing-pod-7f8b9c-ghi on pod C
    // Three different keys per customer → three Stripe charges ch_A, ch_B, ch_C
    val podName = sys.env.getOrElse("HOSTNAME", system.name)
    val key = DigestUtils.sha256Hex(
      s"${customer.id}:${customer.billingPeriod}:$podName"
    ).take(32)

    stripeClient.chargeCustomer(customer, key)
  }
  .runWith(Sink.ignore)

The failure mode is identical to Vert.x’s setPeriodic() cluster broadcast: N pods, N concurrent billing loops, N unique idempotency keys per customer per billing period (because each key includes the pod-specific hostname). The charges appear in Stripe’s dashboard as N legitimate charges for the same customer in the same billing period, spread across the time window where each pod’s billing loop runs.

The failure also occurs with a content-hash key (no pod-specific value) when the timer intervals are not perfectly synchronized. With a 30-day interval and three pods starting at slightly different times — from rolling deploys, pod evictions, or Kubernetes node migrations — pod A’s timer fires at T=0, pod B’s timer fires at T+2h (started 2 hours later due to node migration), and pod C’s timer fires at T+26h (started 26 hours later due to spot instance preemption). Within Stripe’s 24-hour idempotency window: pod A creates ch_A at T=0, pod B’s call at T+2h returns the cached ch_A response (safe). Pod C’s call at T+26h arrives after the 24-hour cache has expired. Stripe has no record of the content-hash key at T+26h; it creates ch_B.

The fix for failure mode 3

Restrict billing tick execution to a single elected node using Akka Cluster Singleton. The ClusterSingletonManager ensures that exactly one instance of the billing actor runs across the entire cluster at any time. When the pod running the singleton fails, Akka Cluster promotes the singleton to the next oldest surviving pod. The new singleton pod runs the billing tick; all other pods run no billing timer at all:

import akka.actor.typed.{ActorSystem, Behavior}
import akka.actor.typed.scaladsl.Behaviors
import akka.cluster.typed.{ClusterSingleton, SingletonActor}
import akka.stream.scaladsl._
import scala.concurrent.duration._

// BillingTick actor — runs as a ClusterSingleton
object BillingTickActor {
  sealed trait Command
  case object RunBillingCycle extends Command

  def apply(): Behavior[Command] = Behaviors.setup { context =>
    implicit val system = context.system

    // Source.tick() runs only inside the singleton actor — only one actor exists in the cluster.
    // When this pod fails, Akka Cluster promotes the singleton to the next pod.
    // The new pod starts its own Source.tick() — with the same billing period, the pre-flight
    // check prevents duplicate charges for customers billed by the previous singleton pod.
    Source
      .tick(initialDelay = 1.minute, interval = 30.days, tick = RunBillingCycle)
      .mapAsync(1) { _ =>
        customerRepo.findAllDueForBilling()
      }
      .mapConcat(identity)
      .mapAsync(4) { customer =>
        // Stable content-hash key — no pod name, no system.name (changes on pod restart),
        // no ActorSystem creation timestamp, no singleton handoff sequence number.
        val key = DigestUtils.sha256Hex(
          s"${customer.id}:${customer.billingPeriod}:akka-billing"
        ).take(32)

        billingRepo.insertIfAbsent(customer.id, customer.billingPeriod, key).flatMap {
          case 0 => Future.successful("already_billed")
          case _ => stripeClient.chargeCustomer(customer, key)
        }
      }
      .runWith(Sink.ignore)

    Behaviors.receiveMessage {
      case RunBillingCycle =>
        // Tick handled by the stream above
        Behaviors.same
    }
  }
}

// Register the singleton in each pod's ActorSystem at startup.
// Only one instance runs across the entire cluster at any time.
val singleton = ClusterSingleton(system)
val billingProxy = singleton.init(
  SingletonActor(BillingTickActor(), "billing-tick-singleton")
)

The ClusterSingleton serializes billing execution across the cluster. When the pod running the singleton fails, Akka Cluster detects the failure within akka.cluster.failure-detector.acceptable-heartbeat-pause (default: 3 seconds after the heartbeat threshold is crossed) and promotes the singleton to the next oldest node. The new singleton starts its billing tick from scratch. The pre-flight database check prevents duplicate charges for customers already billed by the previous singleton pod in the current billing period: the new pod’s pre-flight INSERT ... ON CONFLICT DO NOTHING finds the existing billing record and skips Stripe for customers whose billing record was committed before the previous pod failed.

The singleton handoff window — the time between the previous pod’s failure and the new pod taking over — is typically 5–15 seconds in a well-configured Akka Cluster. Customers whose pre-flight insert committed but whose Stripe call was in flight during this window will have a pending billing record with no charge_id. The reconciliation job detects these records by querying WHERE status = 'pending' AND created_at < now() - INTERVAL '5 minutes' and re-issues stripe.charges.create() with the same content-hash key. Within Stripe’s 24-hour window, this returns the cached ch_A response. Outside the window (unlikely for a 5–15 second handoff, but possible if the reconciliation job itself was delayed), the same content-hash key creates ch_A fresh — identical billing outcome, no duplicate charge.

An alternative to ClusterSingleton for teams already running a database: use a distributed lock via a SELECT ... FOR UPDATE on a billing_periods table row, or a pg_try_advisory_lock() at billing cycle entry. The pod that acquires the lock runs the billing cycle and releases the lock on completion. Other pods attempt the lock, fail immediately (advisory lock, not blocking), and skip the billing cycle. This approach does not require Akka Cluster membership and works identically with Kubernetes standalone pods that are not cluster-aware.

-- Advisory lock approach (no Akka Cluster required)
-- Each pod acquires a per-period advisory lock before running the billing loop.
-- Only one pod acquires it; others skip immediately.

-- Lock ID: hash of the billing period string, cast to bigint for pg_try_advisory_lock.
-- 2026-08 → hashtext('billing:2026-08') → deterministic bigint per period.

SELECT pg_try_advisory_lock(hashtext('billing:' || $1)::bigint);
-- Returns true if this pod acquired the lock, false if another pod holds it.
-- Lock is released at session end (connection close) or explicitly.

-- On completion:
SELECT pg_advisory_unlock(hashtext('billing:' || $1)::bigint);

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

All three failure modes are addressed at the application layer by content-hash idempotency keys and pre-flight database checks. A vault key per billing period provides a hard backstop when the application-layer fix fails to prevent all duplicate paths — two pods acquiring the database advisory lock within the same millisecond (race on lock grant), a pre-flight row written but Stripe call not yet issued when the singleton is promoted to a new pod, or a content-hash key that expires from Stripe’s 24-hour cache due to a delayed singleton handoff.

A vault key is a scoped proxy key issued to the billing system for a specific vendor (Stripe) and a specific billing period (e.g., August 2026). The proxy enforces a spend cap of expected_monthly_revenue × 1.10: 10% above the expected billing total. 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 and the vault key that made it.

// Issue a vault key for this billing period before starting the billing stream.
// Cap = expected total × 1.10 — hard backstop against double-charge scenarios
// that slip through the application-layer idempotency check.

POST https://proxy.keybrake.com/keys
{
  "vendor": "stripe",
  "billing_period": "2026-08",
  "daily_usd_cap": null,
  "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"
}

// Use vault_key as the Authorization bearer in all Stripe calls for this billing period.
// If a duplicate charge would exceed the cap, the proxy rejects it:
// HTTP 429 Too Many Requests — "spend cap exceeded for billing_period 2026-08"
// Audit log records: customer_id, attempted_amount_cents, vault_key, timestamp.

The spend cap closes the failure when the advisory lock race produces two concurrent billing loops. If two pods both acquire the pg_try_advisory_lock within the same millisecond window (possible when both pods start within the same server-side lock grant cycle — a race that is rare but non-zero under load), both run the billing loop concurrently. The content-hash key is identical from both pods (no pod-specific value), so Stripe’s 24-hour cache deduplicates concurrent calls for the same key within the window. But the pre-flight UNIQUE (customer_id, billing_period) constraint serializes concurrent inserts: the first pod’s insert wins; the second pod’s insert returns zero rows affected and skips Stripe. If the second pod’s error handling treats the UNIQUE constraint violation as a retryable error and calls Stripe anyway, the vault key cap stops the duplicate charge at the proxy layer.

Gap analysis

mapAsync supervision strategy: resumingDecider silently drops failed billing records

Akka Streams’ default supervision strategy for a stream failure is Stop: the stream fails and the materialized value future completes with the exception. RestartSource catches this failure and re-materializes the stream. An alternative supervision strategy is resumingDecider or restartingDecider, applied via ActorAttributes.supervisionStrategy(...) on the mapAsync stage. With resumingDecider, a failed future in mapAsync is silently skipped: the stream continues processing the next element without emitting anything for the failed element. A billing record whose Stripe call throws a StripeException is silently dropped. The customer is never billed. No exception propagates to the restart wrapper. The stream continues — no restart, no visibility into the dropped record, no audit log entry.

Do not use resumingDecider on the mapAsync stage in billing streams. Use Stop (the default), which propagates the exception to RestartSource and triggers a restart with the full billing loop from the beginning (protected by the pre-flight check). Or use explicit error handling within the mapAsync future: recover transient StripeExceptions with a delay-and-retry, log permanent failures (e.g., card declines) to a dead-letter queue for manual review, and propagate system-level failures (database connection lost) as stream failures for RestartSource to catch.

Stream graph restart semantics: RestartFlow vs. RestartSource for partial stream resilience

A billing pipeline that reads from Kafka (reliable source) and calls Stripe (potentially failing) can wrap just the Stripe-calling stage in RestartFlow rather than wrapping the entire pipeline in RestartSource. RestartFlow.withBackoff(() => Flow[Customer].mapAsync(N)(chargeCustomer))` restarts only the downstream Stripe-calling flow when it fails, while the Kafka source continues emitting messages. The Kafka consumer continues pulling messages and buffering them upstream during the restart backoff. Messages that were in-flight through the failing flow on failure are replayed when the new flow materializes.