Part 1 of 2: how a small, home-grown data pipeline became the backbone of Tilt’s Kotlin Multiplatform data layer. Part 2 covers what it unlocked: OpenTelemetry tracing with Grafana.
At Tilt, the Android and iOS apps share one Kotlin Multiplatform module that owns networking, caching and business logic. A year ago, every repository function in it was a small, hand-rolled piece of coroutine plumbing. Today almost every new one is a single expression that reads like a sentence:
source(apolloClient.query(ProductQuery(id)))
.transform { converter.toDomain(it) }
.buildOnce()
This article is about the roughly 1,600 lines of Kotlin behind that expression: why I built them instead of adopting a library, the first design I threw away, and the one idea that survived every refactor: everything is a source.
There is also a twist I did not plan for. I built the framework as guardrails for developers who did not know Flows well. It turned out to be exactly the guardrails AI coding agents needed too.
The problem: more contributors, less Flow experience
In late 2025, iOS developers started contributing to the shared module. That was the point of Kotlin Multiplatform, and it was great. But coroutines and Flows have plenty of sharp edges: when a flow completes, what flatMapLatest cancels, where an exception ends up, why you must never swallow a CancellationException. These are not obvious if Swift is your daily language.
The bigger issue was not knowledge of Flows. It was error handling that was quietly skipped. A typical repository function looked like this:
return flow {
emit(executeDataCatching(apolloClient.query(BundlesConfigQuery(roomId))) { /* map */ })
}.flatMapLatest { configResult ->
if (configResult.isSuccess) {
val config = configResult.getOrThrow()
if (config.sellerId != null && config.enabled) {
combineTiltResult(
apolloClient.subscription(BundlesSubscription(config.sellerId)),
apolloClient.subscription(AddressSubscription(roomId)),
) { bundles, address -> converter.toDomain(bundles, address) }
} else {
flowOf(TiltResult.success(ShippingBundleState.Inactive))
}
} else {
flowOf(TiltResult.failure(configResult.exceptionOrThrow()))
}
}.retryUntilChanged(LIVE_MODEL)
It works. But the isSuccess / getOrThrow / flowOf(failure) ladder is written by hand every time. Each time, someone can forget a branch, call getOrThrow() one line too early, or wrap a runCatching around something that should have stayed typed. An audit of the codebase at the time found exactly that: errors wrapped inconsistently, and at least one call that skipped the wrapper altogether.
The second problem was combining. The existing helper, combineTiltResult, only understood Apollo calls, plus a few hand-picked mixes with Flows. Every new combination of arity and input type needed a new overload. There were 11 of them, each repeating the same body. I needed to combine anything: a query with a subscription, a local store, a cached value, a settings flow.
One thing was not up for debate: TiltResult, Tilt’s own result type. It existed before any of this, because Kotlin’s Result does not bridge to Swift. Whatever I built had to speak TiltResult end to end.
Build or adopt?
I did not go shopping for a library. That was a deliberate choice, not an oversight.
What I needed was small: wrap a few kinds of input into one shape, combine them, transform them, and handle errors the same way everywhere. That does not justify an external SDK with its own release cycle, its own opinions about caching, and its own learning curve. I also wanted full control. This would sit under every repository in the module. When it needed to change, I wanted to change it that afternoon, not open an issue upstream.
That decision paid off later in a way I could not have predicted. More on that at the end.
Attempt one: a DSL nobody could discover
The first version, in November 2025, was a builder DSL. It looked tidy on a slide:
dataPipeline {
addSource<UserData>(observe(apolloClient.subscription(UserSubscription())))
addSource<RoomData>(observe(apolloClient.subscription(RoomSubscription(roomId))))
combine<UserData, RoomData> { user, room -> converter.toDomain(room, user) }
}.build()
In practice it was painful to use. IDE auto-completion barely helped: inside the builder lambda, you had to already know what to type. It was not clear from the code how the pieces fit together, or which calls were legal in which order. Under the hood, combine checked at runtime that you had registered the right number of sources, and cast them unchecked. The compiler could not help you either.
There was also a split between fetch(…) for one-shot calls and observe(…) for streams. It sounded clear. But Apollo already emits once for queries and mutations and keeps emitting for subscriptions. The second vocabulary only added a way to pick the wrong one.
In January 2026 I replaced the whole thing. That rewrite is where the real idea showed up.
The idea that stuck: everything is a source
The rewrite dropped the builder and kept one concept: a PipelineSource<T>. Under the hood it is just a Flow<TiltResult<T>>, so a failure is a value that travels down the chain instead of an exception that escapes it. You never touch that flow directly. You build sources, connect them, and close the pipeline with a terminal.
There are three kinds of building blocks.
1. One factory for every input. source(…) has overloads for an Apollo call (query, mutation or subscription), a Flow<T>, a plain value, or a TiltResult you already have. Whether it emits once or many times is a property of the input, not something you declare.
2. Operators, named by intent. transform, transformToSourceLatest for dependent calls, validate, distinctUntilChangedBy, timeout. Error operators follow one naming rule: with* changes the stream (withFallbackValue, withFallbackSource), on* only observes it (onSuccess, onFailure). Every operator catches exceptions and turns them into a typed failure. A checkNotNull inside a transform becomes a parsing error the caller can handle, not a crash.
3. Combining, with types the compiler checks. sourceCombine(a, b, c) { a, b, c -> } is a plain function with fixed arities. It short-circuits on the first failure, so the lambda only runs when every input succeeded and its parameters are never wrapped in a result. For a dynamic number of inputs there is combineSources(list).
Then you pick a terminal. build() returns a Flow<TiltResult<R>> for long-lived streams and adds retry with backoff and de-duplication. buildOnce() is a suspend function returning TiltResult<R> for one-shot reads.
Here is the function from the previous section, rewritten:
val configSource = source(apolloClient.query(BundlesConfigQuery(roomId)))
.transform { data -> data.toConfigParams() }
val bundlesSource = configSource.transformToSourceLatest { config ->
if (config.sellerId != null && config.enabled)
source(apolloClient.subscription(BundlesSubscription(config.sellerId))).asNullable()
else emptySource()
}
val countrySource = source(apolloClient.subscription(AddressSubscription(roomId)))
.transform { it.address?.country }
return sourceCombine(configSource, bundlesSource, countrySource) { config, bundles, country ->
converter.toDomain(bundles, config, country)
}.build()
The ladder is gone. Nobody writes isSuccess or getOrThrow, because short-circuiting on failure is the framework’s job. Each source is a named value you can read in isolation. And all 11 combineTiltResult overloads were eventually deleted.
The idea grew past repositories. The in-memory stores now expose fooSource(): PipelineSource<T> instead of a suspend fun getFoo(). A repository that needs the store’s state just composes it into its own pipeline, with no flow { emit(store.getState()) } trick in between. Before landing there, I considered letting a core operator accept suspend lambdas so the trick would not be needed. I rejected it, because it would invite network calls and delay() inside lambdas instead of modelling them as sources. If something produces data, it is a source.
Failures propagate. Degradation is a decision.
The most important rule in the framework is not in any operator. It is a default: a failure reaches the consumer unless someone decides otherwise.
This sounds obvious until you watch it erode. When one of three combined sources can fail, the easy fix is to wrap each of them in .asNullable().withFallbackValue(null). Nothing blocks, the screen renders, and everyone moves on. But now the data can be stale, and the place that observes it and would log the problem never hears about it. I pushed back on exactly this pattern more than once.
The opposite mistake is just as real. When I migrated the live auction stream, one failed product lookup could suddenly fail the whole live room. Before, it only dropped one auction from the list. That is not acceptable, and it is precisely what fallback operators exist for. So the rule has two halves:
- Propagate by default. A plain source has no fallback.
- Degrade on purpose, where the domain says one part must not take down the whole. Make that decision visible in the code, and log before you swallow the error.
The home feed shows both halves in one function. It fetches two things in parallel: the user’s country from their settings, which only fine-tunes the results, and the feed itself. A slow or failed settings call must never block the feed. A failed feed must reach the caller. Before the migration, that took hand-rolled coroutines:
): TiltResult<ContentGroupPage> = coroutineScope {
val getUserCountryDeferred = async {
withTimeoutOrNull(USER_SETTINGS_TIMEOUT) {
userRepository.getUserSettings()
.onSuccess { logger.i { "User settings: $it" } }
.onFailure { logger.e(it) { "Failed fetching user settings" } }
.getOrNull()?.country
}
}
val feedDeferred = async {
executeDataCatching(apolloClient.query(PersonalFeedQuery(/* … */))) { data -> data }
}
val userCountry = getUserCountryDeferred.await().also { userCountry = it }
return@coroutineScope feedDeferred.await().map { data -> /* convert */ }
}
The degradation is there, but it is buried in withTimeoutOrNull and getOrNull(). Nothing says which of the two fetches is allowed to fail. Here is the same function on the pipeline:
val userCountrySource = userSettingsStore.userSettingsSource()
.onFailure { logger.e(it) { "Failed fetching user settings" } }
.transform { it.country }
.asNullable()
.timeout(USER_SETTINGS_TIMEOUT.inWholeMilliseconds, null)
.withFallbackValue(null) // settings may never block the feed
val feedSource = source(apolloClient.query(PersonalFeedQuery(/* … */)))
return sourceCombine(userCountrySource, feedSource) { country, data ->
/* convert */
}.buildOnce(ShopperTracedOperation.GetFeedContentGroups(/* … */))
Now the decision reads off the code. The settings source degrades on purpose: it logs, times out and falls back to null. The feed source has no fallback, so its failure reaches the caller. And because the settings store exposes a source, it composes like any other input.
One more distinction lives in this rule: absent is not failed. “This user has no active bundle” is a successful response with a null field. “The bundle could not be fetched” is a failure. Mixing those up is how fallbacks start hiding outages.
A migration story: when “the stream is the cache” was too clever
The best test of the framework was the busiest stream in the app: the auctions in a live room. Each auction references a product, so the old code subscribed to auctions, collected product IDs, and fetched missing products into a hand-maintained cache:
private val productsCache = mutableMapOf<String, Product>()
apolloClient.subscription(AuctionsSubscription(roomId))
.mapToTiltResult { auctions ->
val ids = auctions.productIds()
cleanupUnusedProducts(ids)
productsCache.putAll(fetchProductsConcurrently(ids - productsCache.keys)) // async + awaitAll
AuctionsWithProducts(auctions, buildRelevantProductsMap(ids))
}
The first migration looked beautiful. A new combineSources(list) operator fanned out one source per product. distinctUntilChangedBy { ids } skipped work when the product set had not changed. The manual cache and three helper functions were deleted. The cache was no longer a data structure but a live subscription. The stream was the cache.
Then code review found the catch. distinctUntilChangedBy held back the whole stream, so auction updates that did not change the product set never reached the screen. And any change to the set re-fetched every product, not just the new one.
The agent working with me proposed a new generic operator, combineKeyedSources, complete with tests. I stashed it. It solved the problem by adding a framework primitive for one call site. The fix I shipped was smaller: a plain cache that hands out sources.
private fun productSource(id: String): PipelineSource<Product?> {
productsCache[id]?.let { return source(it) } // hit: a value source, no network
return source(apolloClient.query(ProductQuery(id)))
.transform { it.toProduct() }
.onSuccess { productsCache[id] = it }
.onFailure { logger.w(it) { "Failed to fetch product $id" } }
.withFallbackValue(null)
}
A side effect I did not expect: the onFailure line logs the cause of a failed product fetch. The old imperative code used getOrNull() and threw it away silently.
The lesson for me: deleting state in favour of an elegant stream changed behaviour in ways no type caught. The right answer was not a cleverer operator. It was a boring cache that speaks the framework’s language.
Guardrails for people turned out to be guardrails for agents
I designed the framework for a developer who does not know every nuance of Flows and might skip error handling under deadline pressure. A few months later, the team was writing a large share of its code with AI agents. An agent fits that description surprisingly well.
Agents are fluent in coroutines in general and unreliable in the specifics. Left alone, they reach for whatever pattern they saw most often in training. With the framework, the specifics are narrowed down to a small vocabulary: make sources, combine them, end with build() or buildOnce(). I wrote that vocabulary into the agent guides as the mandatory pattern for repository methods, and a pipeline framework guide sits next to the architecture docs.
The mistakes I still catch are telling, because they are almost always a step away from the framework:
- Abandoning sourceCombine for hand-written async/awaitAll, convinced the pipeline broke parallelism. It did not. For one-shot sources, combine runs them concurrently and cancels them together. The real culprit was a test mock server that answered requests in order. The pipeline version was about 10 lines instead of 30.
- Using the wrong terminal, like build().first() instead of buildOnce(). That quietly adds retry and de-duplication meant for long-lived streams.
- Adding fallbacks by reflex, which is the failure-propagation problem from earlier.
Each of these is easy to spot in review because the framework makes the correct shape so recognisable. A deviation stands out. The team took to it well, and with the guides in place, agents built new features on top of the pipeline quickly and consistently. Today more than 50 files in the module build on source(…). Older code still uses the previous helpers, and it moves over function by function, which was a design goal from day one.
What I would tell myself in November 2025
- Discoverability beats elegance. The DSL looked better in a design doc. Free functions on plain values won because your IDE can complete them and a newcomer can read them top to bottom.
- Defaults matter more than features. The operator list grew over time. The thing that actually changed code quality was one default: failures travel to the consumer unless someone opts out, in writing.
- One concept scales further than many. fetch vs observe, builders, special-purpose operators: each was a second way to say the same thing. “Everything is a source” kept absorbing new cases, from stores to caches, without new vocabulary.
- The framework got simpler as it matured. Months later I seriously considered going back to a DSL-first build { } shape. An analysis showed it would change nothing at runtime, touch over 180 call sites, and leave two vocabularies to maintain. I kept the boring version.
- Own your small infrastructure. Because the code is mine, I can change a core operator the day I need it, with no workaround on top of a library.
That last point is where Part 2 begins. This summer I set out to add distributed tracing for business operations like placing a bid, across both apps. In most codebases that means touching every call site. In this one, every business operation already ended in build() or buildOnce(), and every network call already went through source(…). So tracing a function cost one argument:
.buildOnce(ShopperTracedOperation.MakeBid(roomId, auctionId, amount))
In Part 2: how I built mobile OpenTelemetry tracing on top of the pipeline, what Grafana revealed that had been invisible before, and what it is like to run an alpha OpenTelemetry SDK for Kotlin Multiplatform in production.
Everything Is a Source — Part 1 was originally published in ProAndroidDev on Medium, where people are continuing the conversation by highlighting and responding to this story.