Error Handling Across the Bridge: Exceptions, onError, and Fallbacks

Coroutines and Reactor disagree about what an error *is*. In suspending code, failure is an exception that propagates up the call stack and is caught with a familiar try/catch. In Reactor, failure is a value that travels down the pipeline: the reactive stream emits an onError signal, terminates, and that signal is handled with operators like onErrorResume, onErrorReturn, or retry. When you mix the two worlds with kotlinx-coroutines-reactor, the bridge functions translate one representation into the other automatically. Understanding exactly where that translation happens is the difference between robust code and swallowed failures that surface as mysterious 500s in production.

Going from Reactor into a coroutine, awaiting a publisher converts the terminal onError signal back into a thrown exception. When you call mono.awaitSingle(), awaitSingleOrNull(), or flux.asFlow(), an upstream onError is re-thrown at the suspension point, so a plain try/catch around the await call works exactly as your intuition expects. The subtlety is the *empty* completion: awaitSingle() throws NoSuchElementException if the Mono completes without a value, while awaitSingleOrNull() returns null instead. Reaching for the wrong one turns a perfectly normal empty result into an exception you then have to handle, so choose the awaiting function based on whether emptiness is a valid outcome in your domain.

Going the other direction, from a coroutine into Reactor, the mono { } and flux { } builders catch any exception thrown inside the suspending block and convert it into an onError signal on the resulting publisher. This means the exception does not escape as a thrown Kotlin exception to the surrounding code — it becomes a reactive error that the downstream pipeline must handle with onErrorResume and friends. A common mistake is wrapping a mono { } call in try/catch expecting to catch the failure synchronously; the builder returns immediately and the error only materializes when something subscribes. Handle it as a signal on the stream, not as a thrown exception at the call site.

Mapping domain exceptions cleanly is where this pays off. Inside a coroutine you throw rich, typed exceptions (OrderNotFoundException, PaymentDeclinedException); when that coroutine is exposed as a Mono via the builder, those same exceptions ride along as the throwable carried by the onError signal — emitted directly, not wrapped. Downstream you can pattern-match on type with onErrorResume { e -> ... } where e is that exact exception, or, staying in coroutine land, with try/catch and a when (e) block. Keep your exception hierarchy meaningful on the suspend side and you will not need fragile string matching on the Reactor side — the type information survives the crossing intact.

Structured concurrency changes the failure semantics, and this is the part that surprises people. In a normal coroutineScope, if any child coroutine fails, the exception cancels all siblings and propagates to the parent — failure is all-or-nothing, which is usually what you want for a single request. A supervisorScope breaks that coupling: a failing child does not cancel its siblings, so each independent task can fail in isolation. Use coroutineScope when sub-results are interdependent and one failure should abort everything; use supervisorScope when you fan out to several independent calls (three downstream services, say) and want to apply a per-call fallback rather than failing the whole batch.

The coroutine analogue of onErrorResume is simply try/catch returning a default value, and the analogue of onErrorReturn is a catch that yields a constant. Because suspending code is sequential and exception-based, fallbacks read like ordinary Kotlin: wrap the call, catch the specific exception, return the fallback. Be deliberate about which exceptions you swallow — catching CancellationException is a serious bug because it breaks structured concurrency and prevents proper cancellation. Always rethrow CancellationException (or catch only your own domain types) so that timeouts and parent cancellation still work.

For retries and timeouts, prefer the coroutine-native tools when you are on the suspend side: withTimeout / withTimeoutOrNull for deadlines, and a small retry loop or a library helper for backoff, all of which integrate with cancellation. If you are still composing a Reactor pipeline, Reactor's retryWhen and timeout operators remain available and emit onError on exhaustion, which the bridge then converts back to a thrown exception when you await. Avoid stacking both layers blindly — a Reactor-level retry under a coroutine-level timeout can produce confusing interactions, so pick the layer that owns the policy and keep the other thin.

A practical rule of thumb: decide which world owns the error-handling logic for a given piece of code and stay there. If the surrounding code is suspending, await early, convert to exceptions at the boundary, and use try/catch, supervisorScope, and withTimeout. If you are deep inside a Reactor chain, handle signals with onErrorResume and only cross into coroutines once the stream is stable. Mixing both at the same level — try/catch around an unsubscribed Mono, or onErrorResume around an already-thrown exception — is where bugs hide. Translate at the edges, handle in the middle.

Finally, observability survives the bridge only if you let it. Log or attach context at the point where the failure is still typed and meaningful — inside the suspend function or right after awaitSingle — rather than after several conversions have flattened it into a generic ReactiveException. When you must inspect a wrapped cause, walk e.cause, and when you rethrow across the boundary, preserve the original as the cause so stack traces stay useful. The bridge is faithful, but only if you handle the error close to where it was born.

import kotlinx.coroutines.reactor.awaitSingle
import kotlinx.coroutines.reactor.awaitSingleOrNull
import reactor.core.publisher.Mono

class UserRepository {
    fun findById(id: String): Mono<User> = Mono.empty() // pretend DB call
}

suspend fun loadUser(repo: UserRepository, id: String): User? {
    return try {
        // awaitSingleOrNull -> empty completion returns null (no exception);
        // awaitSingle() would instead throw NoSuchElementException on empty.
        repo.findById(id).awaitSingleOrNull()
    } catch (e: DataAccessException) {
        // An upstream onError signal is re-thrown here and caught normally.
        logger.warn("lookup failed for $id", e)
        null
    }
}

import kotlinx.coroutines.reactor.mono
import reactor.core.publisher.Mono

class OrderNotFoundException(id: String) : RuntimeException("order $id not found")

fun orderMono(service: OrderService, id: String): Mono<OrderDto> =
    mono {
        val order = service.fetch(id) // suspend; may throw OrderNotFoundException
        order.toDto()
    }.onErrorResume(OrderNotFoundException::class.java) {
        // 'it' IS the thrown OrderNotFoundException, carried directly by the onError signal.
        Mono.just(OrderDto.empty())
    }

import kotlinx.coroutines.*

suspend fun fetchPrice(source: String): Int {
    if (source == "flaky") throw RuntimeException("$source is down")
    delay(50)
    return source.length * 10
}

suspend fun gatherPrices(): List<Int> = supervisorScope {
    val sources = listOf("alpha", "flaky", "beta")
    val deferreds = sources.map { src ->
        async {
            try {
                fetchPrice(src)
            } catch (e: CancellationException) {
                throw e // never swallow cancellation
            } catch (e: Exception) {
                -1 // onErrorReturn-style fallback, isolated per child
            }
        }
    }
    deferreds.awaitAll()
}

fun main() = runBlocking {
    println(gatherPrices()) // [50, -1, 40]
}

import kotlinx.coroutines.*

suspend fun slowCall(): String {
    delay(2000)
    return "real result"
}

suspend fun resilientFetch(): String =
    withTimeoutOrNull(500) { slowCall() } ?: "cached fallback"

fun main() = runBlocking {
    println(resilientFetch()) // cached fallback (slowCall exceeded 500ms)
}