Flow Context, Errors & Backpressure Explained
KOTLIN › Flow
A Flow has a rule that shapes everything in this topic: **the collector decides the context, and the flow must not change it behind the collector's back.** That rule is called **context preservation**.
// In a ViewModel, on Dispatchers.Main
repo.users().collect { users ->
_state.value = UiState(users) // guaranteed to be on Main
}
Whatever repo.users() does internally, the collect lambda runs in the context that called collect. You never have to wrap it in withContext(Dispatchers.Main) to be safe, and the flow cannot surprise you by delivering on a pool thread.
The library enforces this rather than trusting you. Try to change context inside a flow { } builder and you get a runtime error:
val bad = flow {
withContext(Dispatchers.IO) { emit(load()) }
}
// IllegalStateException: Flow invariant is violated:
// Flow was collected in [Main], but emission happened in [IO]
That message is worth recognising on sight, because the fix is never to add more withContext. The declarative operator flowOn is the supported way to move the producer, and the rest of this topic is about it, about where failures are allowed to be caught, and about what happens when the producer outpaces the collector.
Three concerns, one theme: who is running, and what happens when the two ends disagree.
**flowOn changes only the upstream context, and only upstream.** Anything declared *before* flowOn in the chain runs on the dispatcher you give it; everything *after* it, and the collector itself, stays in whatever context called collect.
flow { emit(loadData()) } // runs on IO
.map { it.transform() } // also runs on IO
.flowOn(Dispatchers.IO) // only affects everything ABOVE this line
.collect { value -> // runs in the CALLER's context, e.g. Main
updateUi(value)
}
This is called **context preservation**, and it's why you never need to wrap collect { } in its own withContext to get back to the main thread: flowOn already guarantees the collector sees whatever context it was called from, untouched.
Two details about flowOn come up once people know the basic rule.
**It inserts a channel.** Moving the producer to another dispatcher means values have to cross a thread boundary, and flowOn does that with a buffered channel. So flowOn implies buffer: the upstream can run ahead of the collector rather than emitting in lockstep. That is usually a benefit, and it is why adding flowOn sometimes changes timing in ways people find surprising.
**Multiple flowOn calls each affect only what is above them.**
flow { emit(readFile()) } // IO
.flowOn(Dispatchers.IO)
.map { parse(it) } // Default
.flowOn(Dispatchers.Default)
.collect { render(it) } // caller's context, e.g. Main
Read it bottom-up: each flowOn claims everything above it that has not already been claimed. This is genuinely useful when a pipeline has a blocking stage and a CPU stage, and it is a good answer to "how would you read a file and parse it off the main thread".
The mistake to avoid is placing flowOn at the wrong end:
flow { emit(readFile()) }
.collect { render(it) }
// .flowOn(...) here would be meaningless: nothing follows a terminal operator
flowOn is an intermediate operator, so it has to sit in the chain, and everything it affects is upstream of where you put it.
Flow also enforces **exception transparency**: a failure from the producer should propagate downstream the same way a value does, and the rule that guarantees this is that emit may only be called directly from the flow { } builder's own coroutine context, never from inside a withContext block.
// BAD: throws "Flow invariant is violated"
val bad = flow {
withContext(Dispatchers.IO) { emit(loadData()) } // throws IllegalStateException
}
// GOOD: keep emit in the builder's own context, shift the whole producer with flowOn
val good = flow {
emit(loadData())
}.flowOn(Dispatchers.IO)
Wrapping emit in your own try/catch inside the builder breaks the same guarantee for a different reason: it hides producer failures from everything downstream instead of letting them propagate to be handled properly. If you genuinely need to push values from a different coroutine or a callback, flow { } isn't the right builder at all, channelFlow or callbackFlow are built for exactly that and don't carry this restriction.
You recover from an upstream failure with the catch operator, never by wrapping emit in a try/catch yourself, that would violate the rule from a moment ago. Inside catch you can log, rethrow, or emit a fallback value to keep the flow alive.
flow { emit(1); throw RuntimeException("upstream") }
.catch { e -> emit(-1) } // catches the upstream exception, emits a fallback
.collect { value -> /* ... */ }
The limitation to hold onto: catch only sees exceptions thrown by the producer and by operators placed above it in the chain. It has no visibility into the collect lambda, or into anything placed below it.
flow { emit(1) }
.catch { println("never reached for collector errors") }
.collect { throw RuntimeException("collector error") } // NOT caught above
To handle a failure thrown inside collect itself, wrap the entire collect call in an ordinary try/catch. catch and try/catch cover two different halves of the same pipeline, upstream and the collector.
catch becomes a real error-handling strategy once you pair it with the two operators either side of it.
**onEach before, catch after** is the shape for a repository exposing a stream to the UI:
val uiState: Flow<UiState> = repo.users()
.map { UiState.Loaded(it) }
.onStart { emit(UiState.Loading) } // before the first upstream value
.catch { e -> emit(UiState.Error(e)) } // upstream failed: emit, don't crash
That is a complete state machine in four lines, and it never throws at the collector. Note the ordering: onStart above catch means a failure inside onStart is also caught.
**retry above catch** handles the transient case first:
repo.users()
.retry(3) { e -> e is IOException } // re-collect upstream on network errors
.catch { e -> emit(UiState.Error(e)) } // only reached once retries are exhausted
retry re-collects the **entire** upstream from scratch, which is worth stating plainly: for a cold flow that means running the producer again, so a flow with side effects will repeat them. retryWhen gives you the attempt index so you can back off:
.retryWhen { cause, attempt ->
if (cause is IOException && attempt < 3) { delay(1000 * (attempt + 1)); true } else false
}
And onCompletion runs on **every** ending, value-exhausted, failed, or cancelled, which makes it the right place for "hide the spinner":
.onCompletion { cause -> _loading.value = false } // cause is null on success
**Backpressure** is the name for what happens when a producer emits faster than a collector consumes, and Flow's answer is different from the one reactive-streams libraries give.
RxJava and Reactive Streams use explicit demand: the subscriber calls request(n) and the producer is obliged not to exceed it, with an overflow strategy for when it does. Flow has no request at all. Instead, **emit is a suspending function**, so if the collector is not ready, the producer simply suspends inside emit until it is.
flow {
repeat(1000) { emit(it) } // each emit suspends until the collector is done
}.collect { slowWork(it) } // producer runs in lockstep with this
That is backpressure for free, with no protocol, no operator, and no overflow strategy to configure. It is also why a plain flow chain never drops values and never grows an unbounded queue: the producer literally cannot run ahead.
The cost is throughput. Lockstep means the producer waits while the collector works, even when they could overlap, and that is exactly what buffer exists to change.
The one place the model leaks is a producer you do not control, a sensor callback or a click listener, which cannot suspend to please you. That is what callbackFlow plus trySend and a BufferOverflow policy are for, and it is the only situation where you have to make an explicit drop-or-suspend decision the way an Rx user always does.
When a producer is fast and the collector is slow, you have three tools, and they behave very differently:
- **buffer**: runs producer and collector concurrently, queueing emissions so nothing is dropped - **conflate**: keeps only the most recent value, silently dropping intermediates the collector couldn't keep up with - **collectLatest**: cancels the in-progress collector block whenever a new value arrives and restarts it fresh
flow { repeat(10) { i -> emit(i); delay(10) } }
.conflate()
.collect { value ->
delay(100) // slow
println(value) // only ever sees the latest, not all 10
}
collectLatest is the one to reach for when the collector block itself does expensive, cancellable work, an in-flight network call, say, and only the newest input actually matters.
buffer takes arguments, and they matter as much as the operator itself.
.buffer(capacity = 64, onBufferOverflow = BufferOverflow.SUSPEND)
The overflow policy decides what happens when the buffer is full:
- SUSPEND (default): the producer waits, so nothing is lost. Ordinary backpressure, with a queue in front of it. - DROP_OLDEST: discard the front of the queue. This is what conflate() is, exactly: buffer(1, DROP_OLDEST). - DROP_LATEST: discard the value being emitted and keep what is queued.
So the three tools you met a moment ago line up on one axis:
| Operator | Producer waits? | Values dropped? | Collector interrupted? | |---|---|---|---| | buffer() | only when full | no | no | | conflate() | never | yes, intermediate ones | no | | collectLatest | never | no (delivered then abandoned) | yes, cancelled and restarted |
That last column is the distinction people miss. conflate never interrupts your collector: a slow collect block runs to completion and simply misses values that arrived meanwhile. collectLatest does interrupt it, cancelling the block mid-flight.
// conflate: each println completes, some numbers never appear
.conflate().collect { delay(100); println(it) }
// collectLatest: printlns are abandoned partway, only the last one completes
.collectLatest { delay(100); println(it) }
Pick conflate when the collector work is cheap and atomic, collectLatest when it is expensive and cancellable, such as an in-flight network call for a search query.
Adding buffer and flowOn to a chain sounds like it should add a queue per operator. It does not, because of **operator fusion**.
flow { ... }
.buffer(10)
.flowOn(Dispatchers.IO)
.buffer(20)
.flowOn(Dispatchers.Default)
flowOn, buffer, conflate and channelFlow all implement an internal interface that lets adjacent instances **merge into a single channel** with a combined configuration, rather than each allocating its own. The library does this because these are exactly the operators people stack up, and a channel per operator would be a real cost.
Two consequences worth knowing:
.buffer(10).buffer(20) // fuses to a single buffer of 20, not two chained buffers
.conflate().buffer(50) // the later buffer wins: conflation is replaced
The second is the one that catches people. conflate() is buffer(1, DROP_OLDEST), so putting a plain buffer after it fuses into one configuration and the conflating behaviour disappears. If you want both a large queue and conflation you have to say so explicitly with buffer(50, BufferOverflow.DROP_OLDEST).
The practical advice: treat buffer, conflate and flowOn as **one channel configuration** for the whole adjacent group, not as independent stages. If you need two genuinely separate buffering behaviours, you need something non-fusable between them.
Cancellation inside a flow follows the same cooperative rules as any coroutine, with one Flow-specific convenience and one Flow-specific gap.
**The convenience:** emit checks for cancellation. Every flow built with the flow { } builder is therefore cancellable at each emission with no effort from you:
val job = launch {
flow { repeat(1000) { emit(it) } }.collect { println(it) }
}
job.cancel() // stops at the next emit
**The gap:** flows that do not go through the builder do not get that check. The classic case is asFlow() on a range or collection, which emits without suspending:
(1..1000).asFlow()
.collect { heavyCpuWork(it) } // no cancellation check anywhere
cancellable() inserts one:
(1..1000).asFlow().cancellable().collect { heavyCpuWork(it) }
And inside a long-running collector block, the ordinary tools apply: ensureActive() to bail out, or yield() to be fair as well as cancellable.
Where this bites in practice is a collect that never ends. Collecting a hot flow, a StateFlow or a database query, suspends forever by design, so the surrounding coroutine only ends when someone cancels it. That is not a bug; it is why the scope you collect in has to be the one whose lifetime you actually want, which is what the sharing and lifecycle topic is about.
Pulling the three concerns together, because interviewers usually probe them as one question about a real pipeline.
val uiState: Flow<UiState> = repo.observeUsers() // cold, hits the database
.map { it.toUiModel() } // CPU work
.flowOn(Dispatchers.Default) // claims both stages above
.onStart { emit(UiState.Loading) }
.retry(2) { it is IOException }
.catch { emit(UiState.Error(it)) } // last, so it covers everything above
.conflate() // UI only needs the newest state
Reading it as an interviewer would:
- **Context:** flowOn claims the database read and the mapping. The collector still runs wherever it was called from, so a ViewModel collecting on Main is safe with nothing extra. - **Errors:** retry handles the transient case by re-collecting upstream, then catch converts what survives into a state rather than a crash. catch is last because it only sees what is above it. - **Backpressure:** emit suspending is the default, so nothing is lost; conflate opts out of that for a UI that only cares about the latest value.
The three failure modes to be able to name:
1. withContext inside flow { } gives "Flow invariant is violated". Use flowOn. 2. catch placed above an operator does not see that operator's failures, and never sees the collector's. Put it last, and use try/catch around collect for the collector. 3. conflate followed by buffer fuses, silently discarding the conflation.