Type-safe async multi-service orchestration using flat chains and visible phases; compiler-enforced argument order, partial-failure tolerance, race/quorum primitives, timeout-fallbacks, memoization and composable resilience.
Type-safe coroutine orchestration for Kotlin Multiplatform.
The code reads like a diagram. The compiler won't let you wire it wrong.
Get Started · Docs · Cookbook
A checkout that calls 7 services in 4 phases. Stock validation needs retry. Payment needs a circuit breaker. Promos have a timeout. Read this and tell me: where are the phases? Where's the retry logic vs the business logic?
The phases are buried. The retry loop broke the async/await rhythm. The circuit breaker is interleaved with the business flow. Add one more concern and it gets worse.
Same 7 calls, same 4 phases, same retry, same circuit breaker, same timeouts. .with = parallel, .then = barrier. Resilience is per-call, inline, composable.
@KapTypeSafe generates a scoped wrapper with per-slot tag interfaces. Each .with { field from value } lambda exposes the slot for the field at the current curry position — the IDE shows exactly the field you owe, by name. Swap two fields and you get a compile error naming the expected tag. Reads like prose (user from fetchUser()), no method-name explosion, single .with/.then for the whole graph.
When
valueis itself aKap<T>(e.g.Kap { fetchX() }.timeout(...)), use parens:.with(field from Kap { ... }). Eitherimport MyTypeKap.*at the top of the file, or qualify inline asMyTypeKap.field. Inside the braces form the lambda's implicit receiver brings tags into scope — no import needed there.
Most use cases don't need retry or circuit breakers. At its core, KAP is just a clean way to run things in parallel and combine the results:
@KapTypeSafe
data class Dashboard(val user: String, val feed: String, val notifications: Int)
kap(::Dashboard)
.with { user from fetchUser() } // ┐
.with { feed from fetchFeed() } // ├─ all three run in parallel
.with { notifications from countUnread() } // ┘
.evalGraph()
Need one field to wait for another? Change .with to .then:
@KapTypeSafe
data class ProfilePage(val user: String, val avatar: String, val recommendations: String)
kap(::ProfilePage)
.with { user from fetchUser(id) } // ┐ parallel
.with { avatar from fetchAvatar(id) } // ┘
.then { recommendations from fetchRecommendations(user) } // waits for user, then fetches
.evalGraph()
Need the result of one graph to decide what to build next? Use .andThen:
kap(::Dashboard)
.with { user from fetchUser() }
.with { feed from fetchFeed() }
.with { notifications from countUnread() }
.andThen { result ->
// use result.user to decide what's next
kap(::FullPage)
.with { dashboard from result }
.with { suggestions from fetchSuggestions(result.user) }
}.evalGraph()
Nothing runs until .evalGraph(). The graph is data — you can build it dynamically, pass it around, compose it.
11 microservice calls. 5 phases. 8 Strings, 2 Booleans, 3 Doubles — the compiler catches every swap:
By default, if any .with branch fails, the whole graph is cancelled — structured concurrency. But sometimes you want partial failures. settled { } wraps the result in Result<A> so a failure doesn't kill the rest:
@KapTypeSafe
data class HomePage(val profile: String, val feed: Result<String>, val ads: Result<String>)
import HomePageKap.*
kap(::HomePage)
.with { profile from fetchProfile() } // critical — failure cancels everything
.with(feed from settled { fetchFeed() }) // optional — failure returns Result.failure
.with(ads from settled { fetchAds() }) // optional — failure returns Result.failure
.evalGraph()
// Feed throws? Profile and ads still complete. You get Result.failure for feed.
Need ALL results even if some fail? sequenceSettled runs every item and collects outcomes:
val results = listOf("svc-a", "svc-b", "svc-c").traverseSettled { svc ->
Kap { callService(svc) }
}.evalGraph()
// → [Success("ok"), Failure(TimeoutException), Success("ok")]
Each one is a method call. No boilerplate, no manual state.
Race — fastest wins, losers cancelled:
val price = raceN(
Kap { pricingFromUS(item) },
Kap { pricingFromEU(item) },
Kap { pricingFromAsia(item) },
).evalGraph() // fastest response wins, other two cancelled
Bounded concurrency — process N items, max M at a time:
val users = userIds.traverse(concurrency = 5) { id ->
Kap { fetchUser(id) }
}.evalGraph() // 100 users, 5 at a time, results in order
Timeout with parallel fallback — don't wait, race against the cache:
val data = Kap { fetchFromApi() }
.timeoutRace(2.seconds, fallback = Kap { readFromCache() })
.evalGraph() // both start at t=0, API wins if fast enough, cache if not
Composable retry — define once, reuse everywhere:
val policy = Schedule.exponential<Throwable>(100.milliseconds)
.jittered() // ±50% random spread (no thundering herd)
.and(Schedule.times(3)) // max 3 attempts
.withMaxDuration(10.seconds) // total budget
Kap { callFlakyService() }.retry(policy).evalGraph()
Timed — measure any call without manual instrumentation:
@KapTypeSafe
data class Dashboard(val user: String, val latency: TimedResult<String>)
import DashboardKap.*
kap(::Dashboard)
.with { user from fetchUser() }
.with(latency from timed { fetchSlowService() }) // TimedResult(value, duration)
.evalGraph()
// dashboard.latency.duration → 230.ms
Memoize — compute once, cache thread-safely:
val config = Kap { loadRemoteConfig() }.memoizeOnSuccess()
// call it from 10 coroutines — only one HTTP request, others wait for the cached result
// if it fails, next caller retries (failures are NOT cached)
Real HTTP — not just simulated delays. This hits the GitHub API:
@KapTypeSafe
data class DeveloperProfile(val user: GithubUser, val topRepos: List<GithubRepo>, val funFact: String)
val profile = kap(::DeveloperProfile)
.with { user from client.get("https://api.github.com/users/kotlin").body() }
.with { topRepos from client.get("https://api.github.com/users/kotlin/repos?sort=stars").body() }
.with { funFact from client.().body<CatFact>().fact }
.evalGraph()
A real order placement. Read it top to bottom — each comment tells you what that line does and why:
Eight production concerns in one function: parallel validation, error accumulation, racing, retry with backoff, circuit breaker, timeout, partial failure tolerance, transactional safety.
KAP is not a replacement for Arrow. They solve different problems and work together:
Either, NonEmptyList, Option, functional error handlingKAP's kap-arrow module builds on Arrow's Either and NonEmptyList for parallel validation with error accumulation. Use both — Arrow for your domain types, KAP for wiring them together:
// Arrow types + KAP orchestration
val result: Either<Nel<OrderError>, OrderResult> =
kapV<OrderError, ValidEmail, ValidAge, User>(::User)
.withV { validateEmail(input) } // returns Either<Nel<OrderError>, ValidEmail>
.withV { validateAge(input) } // returns Either<Nel<OrderError>, ValidAge>
.evalGraph()
// Both run in parallel. Both errors collected. Arrow types, KAP execution.
KAP is just suspend functions in, result out. It works anywhere coroutines work:
No framework, no runtime, no annotation processing at runtime. Your suspend functions go in, your data class comes out.
@KapTypeSafe works on functions too. Same kap(::name).with { … } shape — useful when you want a factory or a transform in front of the constructor (validation, normalization, joining multiple sources into one shape):
@KapTypeSafe
fun buildCheckout: Checkout = Checkout(
user = user.trim(),
cart = cart,
promos = promos.ifBlank { },
stock = stock,
total = shipping + tax,
payment = payment,
)
kap(::buildCheckout)
.with { user from fetchUser() }
.with { cart from fetchCart() }
.with { promos from fetchPromos() }
.then { stock from validateStock() }
.with { shipping from calcShipping() }
.with { tax from calcTax() }
.then { payment from reservePayment() }
.evalGraph()
Constructor reference (::Checkout) and function reference (::buildCheckout) are both first-class — pick whichever matches your domain.
If both your data class and a pass-through factory share the same
(params) -> Rsignature, only annotate one — otherwise both generatefun kap(f: (...) -> R)overloads with identical signatures and the call site is ambiguous.
plugins {
id("com.google.devtools.ksp") // only needed for @KapTypeSafe
}
dependencies {
implementation("io.github.damian-rafael-lattenero:kap-core:3.0.0")
// optional — add any combination
implementation("io.github.damian-rafael-lattenero:kap-resilience:3.0.0")
implementation("io.github.damian-rafael-lattenero:kap-arrow:3.0.0")
implementation("io.github.damian-rafael-lattenero:kap-ksp-annotations:3.0.0")
ksp("io.github.damian-rafael-lattenero:kap-ksp:3.0.0")
}
Starter project — clone and run in 30 seconds.
Use it when:
Skip it when:
coroutineScope { async {} } is fine| Dimension | Raw Coroutines | KAP |
|---|---|---|
| Framework overhead (arity 3) |
KAP adds zero measurable overhead. The abstraction compiles away. Full benchmark suite (119 JMH benchmarks).
900+ tests · Maven Central · Kotlin Multiplatform (JVM, JS, WASM, Native) · Apache 2.0
Get Started · Docs · Cookbook
Surfaced from shared tags and platforms — no rankings paid for.
data class CheckoutResult(
val user: String, val cart: String, val promos: String,
val stock: Boolean,
val shipping: Double, val tax: Double,
val payment: String,
)
val breaker = CircuitBreaker(maxFailures = 5, resetTimeout = 30.seconds)
coroutineScope {
val dUser = async { fetchUser() }
val dCart = async { fetchCart() }
val dPromos = async {
withTimeout(3.seconds) { fetchPromos() }
}
val user = dUser.await()
val cart = dCart.await()
val promos = dPromos.await()
// --- phase 2: validate stock (with retry) ---
var stock = false
var attempt = 0
var delay = 100.milliseconds
while (true) {
try {
stock = validateStock()
break
} catch (e: CancellationException) { throw e }
catch (e: Exception) {
if (++attempt >= 3) throw e
delay(delay)
delay *= 2
}
}
// --- phase 3: shipping & tax in parallel ---
val dShipping = async { calcShipping() }
val dTax = async { calcTax() }
val shipping = dShipping.await()
val tax = dTax.await()
// --- phase 4: payment (with circuit breaker) ---
val payment = if (!breaker.shouldAttempt()) {
throw CircuitBreakerOpenException()
} else {
try {
val p = withTimeout(5.seconds) { reservePayment() }
breaker.recordSuccess()
p
} catch (e: CancellationException) { throw e }
catch (e: Exception) {
breaker.recordFailure()
throw e
}
}
CheckoutResult(user, cart, promos, stock, shipping, tax, payment)
}
data class CheckoutResult(
val user: String, val cart: String, val promos: String,
val stock: Boolean,
val shipping: Double, val tax: Double,
val payment: String,
)
import CheckoutResultKap.*
val retryPolicy = Schedule.exponential<Throwable>(100.milliseconds) and Schedule.times(3)
val breaker = CircuitBreaker(maxFailures = 5, resetTimeout = 30.seconds)
kap(::CheckoutResult)
.with { user from fetchUser() } // ┐
.with { cart from fetchCart() } // ├─ phase 1: parallel
.with(promos from Kap { fetchPromos() }.timeout(3.seconds)) // ┘ + timeout on promos
.then(stock from Kap { validateStock() }.retry(retryPolicy)) // ── phase 2: barrier + retry
.with { shipping from calcShipping() } // ┐ phase 3: parallel
.with { tax from calcTax() } // ┘
.then(payment from Kap { reservePayment() } // ── phase 4: barrier
.withCircuitBreaker(breaker) // + circuit breaker
.timeout(5.seconds))
.evalGraph()
| You write |
|---|
| What happens |
|---|
| Think of it as |
|---|
.with { field from value } | Runs in parallel with everything else in the same phase | "and at the same time..." |
.then { field from value } | Waits for all above, then continues | "once that's done..." |
.andThen { result -> } | Waits, passes the result, builds the next graph | "using what we got..." |
.with { field from ... }
data class CheckoutResult(
val user: String, val cart: String,
val promos: String, val inventory: Boolean,
val stock: Boolean,
val shipping: Double, val tax: Double, val discounts: Double,
val payment: String,
val confirmation: String, val email: String,
)
kap(::CheckoutResult)
.with { user from fetchUser() } // ┐
.with { cart from fetchCart() } // ├─ phase 1: parallel
.with { promos from fetchPromos() } // │
.with { inventory from fetchInventory() } // ┘
.then { stock from validateStock() } // ── phase 2: barrier
.with { shipping from calcShipping() } // ┐
.with { tax from calcTax() } // ├─ phase 3: parallel
.with { discounts from calcDiscounts() } // ┘
.then { payment from reservePayment() } // ── phase 4: barrier
.with { confirmation from generateConfirmation() } // ┐ phase 5
.with { email from sendEmail() } // ┘
.evalGraph()
val checkout = coroutineScope {
val dUser = async { fetchUser() }
val dCart = async { fetchCart() }
val dPromos = async { fetchPromos() }
val dInventory = async { fetchInventory() }
val user = dUser.await()
val cart = dCart.await()
val promos = dPromos.await()
val inventory = dInventory.await()
val stock = validateStock()
val dShipping = async { calcShipping() }
val dTax = async { calcTax() }
val dDiscounts = async { calcDiscounts() }
val shipping = dShipping.await()
val tax = dTax.await()
val discounts = dDiscounts.await()
val payment = reservePayment()
val dConfirmation = async { generateConfirmation() }
val dEmail = async { sendEmail() }
CheckoutResult(
user, cart, promos, inventory, stock,
shipping, tax, discounts, payment,
dConfirmation.await(), dEmail.await()
)
}
data class OrderResult(
val finalPrice: Double,
val reservationId: String,
val paymentId: String,
val notifications: List<Result<Unit>>,
)
import OrderResultKap.*
// Retry: exponential backoff (100ms, 200ms, 400ms) + jitter, max 3 attempts
val retryPolicy = Schedule.exponential<Throwable>(100.milliseconds).jittered() and Schedule.times(3)
// Circuit breaker: after 5 failures, stop calling payment for 30s
val paymentBreaker = CircuitBreaker(maxFailures = 5, resetTimeout = 30.seconds)
suspend fun placeOrder(input: OrderInput): Either<Nel<OrderError>, OrderResult> {
// ── PHASE 1: Validate in parallel, collect ALL errors ────────
val validated = kapV<OrderError, ValidAddress, ValidCard, ValidItems, ValidOrder>(::ValidOrder)
.withV { validateAddress(input.address) } // ┐ all three run at the same time
.withV { validatePaymentInfo(input.card) } // ├─ if address AND card fail,
.withV { validateItems(input.items) } // ┘ you get BOTH errors
.evalGraph()
val order = validated.getOrElse { return Either.Left(it) }
// ── PHASES 2–5: inside DB transaction (rollback on any failure) ──
return bracketCase(
acquire = { db.beginTransaction() },
use = { tx ->
kap(::OrderResult)
// PHASE 2: race 3 pricing providers — fastest wins, losers cancelled
.with(finalPrice from raceN(
Kap { pricingServiceA(order) },
Kap { pricingServiceB(order) },
Kap { pricingServiceC(order) },
))
// PHASE 3: reserve inventory — retry with backoff if flaky
.then(reservationId from
Kap { reserveInventory(tx, order) }
.retry(retryPolicy)
)
// PHASE 4: charge payment — circuit breaker + 5s timeout
.then(paymentId from
Kap { chargePayment(tx, order) }
.withCircuitBreaker(paymentBreaker)
.timeout(5.seconds)
)
// PHASE 5: notifications — if one fails, others still complete
.with(notifications from listOf(
Kap { sendEmail(order) },
Kap { sendPush(order) },
Kap { updateAnalytics(order) },
).sequenceSettled())
.map { Either.Right(it) }
},
release = { tx, exit -> when (exit) {
is ExitCase.Completed -> tx.commit() // success → commit
else -> tx.rollback() // failure or cancel → rollback
}}
).evalGraph()
}
val paymentBreaker = CircuitBreaker(maxFailures = 5, resetTimeout = 30.seconds)
suspend fun placeOrder(input: OrderInput): Either<Nel<OrderError>, OrderResult> {
// ── Phase 1: validate (need supervisorScope to not short-circuit) ──
val (addr, card, items) = supervisorScope {
val dAddr = async { validateAddress(input.address) }
val dCard = async { validatePaymentInfo(input.card) }
val dItems = async { validateItems(input.items) }
Triple(dAddr.await(), dCard.await(), dItems.await())
}
val errors = buildList {
addr.leftOrNull()?.let(::addAll)
card.leftOrNull()?.let(::addAll)
items.leftOrNull()?.let(::addAll)
}
if (errors.isNotEmpty()) return Either.Left(NonEmptyList.fromListUnsafe(errors))
val validAddress = addr.getOrNull()!!
val validCard = card.getOrNull()!!
val validItems = items.getOrNull()!!
val order = ValidOrder(validAddress, validCard, validItems)
// ── Phase 2: race 3 pricing providers (manual select) ──
val bestPrice = coroutineScope {
val jobs = listOf(
async { pricingServiceA(order) },
async { pricingServiceB(order) },
async { pricingServiceC(order) },
)
select {
jobs.forEach { d ->
d.onAwait { price ->
jobs.forEach { it.cancel() }
price
}
}
}
}
// ── Phases 3–5: inside DB transaction (manual try/finally) ──
val tx = db.beginTransaction()
try {
// Phase 3: reserve inventory (manual retry + exponential backoff)
var reservationId: String? = null
var attempt = 0
var retryDelay = 100.milliseconds
while (reservationId == null) {
try {
reservationId = reserveInventory(tx, order)
} catch (e: CancellationException) { throw e }
catch (e: Exception) {
if (++attempt >= 3) throw e
delay(retryDelay)
retryDelay = (retryDelay * 2).coerceAtMost(1600.milliseconds)
// (no jitter — you'd need Random + math here too)
}
}
// Phase 4: charge payment (manual circuit breaker + timeout)
val paymentId = if (!paymentBreaker.shouldAttempt()) {
throw CircuitBreakerOpenException()
} else {
try {
val p = withTimeout(5.seconds) { chargePayment(tx, order) }
paymentBreaker.recordSuccess()
p
} catch (e: CancellationException) { throw e }
catch (e: Exception) {
paymentBreaker.recordFailure()
throw e
}
}
// Phase 5: send notifications (need supervisorScope so one failure doesn't kill the rest)
val notifications = supervisorScope {
listOf(
async { runCatching { sendEmail(order) } },
async { runCatching { sendPush(order) } },
async { runCatching { updateAnalytics(order) } },
).map { it.await() }
}
tx.commit()
Either.Right(OrderResult(bestPrice, reservationId, paymentId, notifications))
} catch (e: Exception) {
tx.rollback()
throw e
}
}
| Module | What it gives you | When you need it |
|---|
| kap-core | .with, .then, .andThen, race, traverse, memoize, settled | Always |
| kap-ksp | @KapTypeSafe, @KapBridge — compile-time named builders with IDE autocomplete | When you want swap-proof, guided orchestration |
| kap-resilience | retry, Schedule, CircuitBreaker, Resource, bracket, raceQuorum, timeoutRace | When you call external services |
| kap-arrow | .withV, kapV, error accumulation with Either<NonEmptyList<E>, A> | When you need parallel validation |
| kap-ktor | respondAsync, KapTracer, circuit breaker plugin | When you use Ktor |
| kap-kotest | shouldSucceedWith, shouldFailWith, shouldCompleteWithin | When you test Kap graphs |
| Arrow | KAP | Together |
|---|
| Parallel execution | parZip (max 9 args) | .with / kap(::T) (unlimited) | KAP orchestrates, Arrow types flow through |
| Error accumulation | zipOrAccumulate | kapV + .withV (max 22 args) | KAP runs validators in parallel, Arrow collects errors |
| Retry / Circuit breaker | Schedule (Arrow) | Schedule + CircuitBreaker (KAP) | KAP's compose inline with .with / .then chains |
| Phase barriers | Manual nesting | .then / .andThen | Only KAP has first-class phases |
// Ktor
get("/checkout/{id}") {
val id = call.parameters["id"]!!
val result = kap(::CheckoutResult)
.with { user from userService.fetch(id) }
.with { cart from cartService.fetch(id) }
.with { promos from promoService.fetch(id) }
.evalGraph()
call.respond(result)
}
// Spring Boot
class CheckoutController(val userService: UserService, val cartService: CartService) {
suspend fun checkout( id: String): CheckoutResult =
kap(::CheckoutResult)
.with { user from userService.fetch(id) }
.with { cart from cartService.fetch(id) }
.with { promos from promoService.fetch(id) }
.evalGraph()
}
// Android ViewModel
import HomeDataKap.*
class HomeViewModel : ViewModel() {
val state = MutableStateFlow<HomeState>(Loading)
fun load() = viewModelScope.launch {
val home = kap(::HomeData)
.with { profile from repo.fetchProfile() }
.with(feed from settled { repo.fetchFeed() })
.with { notifications from repo.countUnread() }
.evalGraph()
state.value = Ready(home)
}
}
| <0.01ms |
| <0.01ms |
| Multi-phase (9 calls, 4 phases) | 180.85ms | 180.98ms |
| 5 parallel calls @ 50ms each | 50.27ms | 50.31ms |