Address PR review comments

Signed-off-by: Matt Ramotar <matt.ramotar@uber.com>

Author: matt-ramotar

store/src/commonMain/kotlin/org/mobilenativefoundation/store/store5/impl/FetcherController.kt

@@ -15,6 +15,7 @@
  */
 package org.mobilenativefoundation.store.store5.impl
 
+import kotlinx.coroutines.CancellationException
 import kotlinx.coroutines.CoroutineScope
 import kotlinx.coroutines.NonCancellable
 import kotlinx.coroutines.async
@@ -86,6 +87,8 @@ internal class FetcherController<Key : Any, Network : Any, Output : Any, Local :
                                             network,
                                             origin = StoreReadResponseOrigin.Fetcher(it.origin),
                                         ) as StoreReadResponse<Network>
+                                    } catch (exception: CancellationException) {
+                                        throw exception
                                     } catch (exception: Throwable) {
                                         StoreReadResponse.Error.Exception(
                                             exception,

store/src/commonMain/kotlin/org/mobilenativefoundation/store/store5/impl/RealStore.kt

@@ -223,6 +223,8 @@ internal class RealStore<Key : Any, Network : Any, Output : Any, Local : Any>(
             }
 
         val requestKeyToFetcherName: MutableMap<Key, String?> = mutableMapOf()
+        // Track if network errored AND this is a fresh request where fallback behavior matters
+        var networkErrorWithNoFallback = false
         // we use a merge implementation that gives the source of the flow so that we can decide
         // based on that.
         return networkFlow.merge(diskFlow).transform {
@@ -233,6 +235,11 @@ internal class RealStore<Key : Any, Network : Any, Output : Any, Local : Any>(
                     val responseOrigin = it.value.origin as StoreReadResponseOrigin.Fetcher
                     requestKeyToFetcherName[request.key] = responseOrigin.name
 
+                    // Track if network errored and fallback to disk is disabled for fresh requests
+                    if (it.value is StoreReadResponse.Error && skipDiskCache && !request.fallBackToSourceOfTruth) {
+                        networkErrorWithNoFallback = true
+                    }
+
                     if (it.value is StoreReadResponse.Data ||
                         it.value is StoreReadResponse.NoNewData ||
                         it.value is StoreReadResponse.Error
@@ -254,6 +261,11 @@ internal class RealStore<Key : Any, Network : Any, Output : Any, Local : Any>(
                     // right, that is data from disk
                     when (val diskData = it.value) {
                         is StoreReadResponse.Data -> {
+                            // Skip disk data if this was a fresh request that errored with fallback disabled
+                            if (networkErrorWithNoFallback) {
+                                return@transform
+                            }
+
                             val responseOriginWithFetcherName =
                                 diskData.origin.let { origin ->
                                     if (origin is StoreReadResponseOrigin.Fetcher) {

store/src/commonTest/kotlin/org/mobilenativefoundation/store/store5/FlowStoreTests.kt

@@ -1207,4 +1207,82 @@ class FlowStoreTests {
                 cancelAndIgnoreRemainingEvents()
             }
         }
+
+    @Test
+    fun stream_givenConverterThrowsWithFallbackDisabled_thenDiskDataNotEmitted() =
+        testScope.runTest {
+            // Given: Pre-populate disk with data, then request fresh with fallBackToSourceOfTruth=false
+            val exception = IllegalStateException("Converter failed")
+            val persister = InMemoryPersister<Int, String>()
+            persister.write(1, "cached value")
+
+            val pipeline =
+                StoreBuilder.from(
+                    fetcher = Fetcher.of { _: Int -> "network" },
+                    sourceOfTruth = persister.asSourceOfTruth(),
+                    converter =
+                        object : Converter<String, String, String> {
+                            override fun fromNetworkToLocal(network: String): String {
+                                throw exception
+                            }
+
+                            override fun fromOutputToLocal(output: String): String = output
+                        },
+                ).buildWithTestScope()
+
+            // When: Request with fallBackToSourceOfTruth=false
+            pipeline.stream(StoreReadRequest.fresh(1, fallBackToSourceOfTruth = false)).test {
+                assertEquals(
+                    Loading(origin = StoreReadResponseOrigin.Fetcher()),
+                    awaitItem(),
+                )
+
+                // Then: Only error is emitted, no disk data (since fallback is disabled)
+                val errorResponse = awaitItem()
+                assertIs<StoreReadResponse.Error.Exception>(errorResponse)
+                assertEquals(exception.message, errorResponse.error.message)
+                cancelAndIgnoreRemainingEvents()
+            }
+        }
+
+    @Test
+    fun stream_givenConverterThrowsWithFallbackEnabled_thenDiskDataEmitted() =
+        testScope.runTest {
+            // Given: Pre-populate disk with data, then request with fallBackToSourceOfTruth=true
+            val exception = IllegalStateException("Converter failed")
+            val persister = InMemoryPersister<Int, String>()
+            persister.write(1, "cached value")
+
+            val pipeline =
+                StoreBuilder.from(
+                    fetcher = Fetcher.of { _: Int -> "network" },
+                    sourceOfTruth = persister.asSourceOfTruth(),
+                    converter =
+                        object : Converter<String, String, String> {
+                            override fun fromNetworkToLocal(network: String): String {
+                                throw exception
+                            }
+
+                            override fun fromOutputToLocal(output: String): String = output
+                        },
+                ).buildWithTestScope()
+
+            // When: Request with fallBackToSourceOfTruth=true
+            pipeline.stream(StoreReadRequest.fresh(1, fallBackToSourceOfTruth = true)).test {
+                assertEquals(
+                    Loading(origin = StoreReadResponseOrigin.Fetcher()),
+                    awaitItem(),
+                )
+
+                // Then: Error is emitted
+                val errorResponse = awaitItem()
+                assertIs<StoreReadResponse.Error.Exception>(errorResponse)
+
+                // And: Disk data is also emitted (since fallback is enabled)
+                val diskData = awaitItem()
+                assertIs<StoreReadResponse.Data<String>>(diskData)
+                assertEquals("cached value", diskData.value)
+                cancelAndIgnoreRemainingEvents()
+            }
+        }
 }