[BUG] Downstream flows never complete for Fetcher Flows that don't emit anything

[hansenji]

Describe the bug
When a fetcher flow just completes without emitting any data the downstream flows will never complete. SourceOfTruth reader won't be called

To Reproduce
fetcher = { key -> flow {} }

Expected behavior
Store should detect when the fetcher flow completes and allow the rest of the process to continue as normal.

Store Version 4.0.0-alpha06

Additional context
Fetchers flows that don't emit anything are extremely useful when manually dealing with caches that can't utilize the OkHttp cache.

[digitalbuddha]

Are you using a source of truth?

[hansenji]

Yes I am using source of truth.

[alexandercasal]

I'm seeing this same issue. I use the StoreResponse to detect if the flow is loading from the network. If so, then I show a loading indicator on the UI.

This works great when I use a nonFlowFetcher, but if I use my own flow that only emits when my cache has expired then I see this sequence:

Cache
Source of truth
Fetcher - > Loading

Since the fetcher remains in a loading state, despite the operation finishing, the loading ui will never disappear.

Here's a sample:

private val myStore = StoreBuilder.from<Key, Dto, MyData>(
        fetcher = fetcher { (key, force) -> fetchIfNeeded(key, force) },
        sourceOfTruth = SourceOfTruth.from(
            reader = { (key, _) ->...... },
            writer = { (key, _), dto ->....... }
        )
    ).build()

private fun fetchIfNeeded(key: String, force: Boolean): Flow<FetcherResult<Dto>> = flow {
        if (force || databaseCacheExpired(key)) {     
             emit(remoteDataSource.fetch(key))
        } 
    }

Expected: Fetcher to not appear to be in a never ending loading state when the flow is finished without emitting. Source of truth should load.

[yigit]

What is the StoreRequest you are passing?

Because if you are passing StoreRequest.fresh, then it would be invalid to fall back to a local value as it is not fresh :/.

Maybe we need to start modeling a StoreResponse.Empty because there is no way to move from Loading to Data without having data :/.

[hansenji]

I am using StoreRequest.fresh.

[yigit]

The fix for this is actually easy, i prototyped it here:
https://github.com/dropbox/Store/tree/yigit/never-emitting-fetcher-185

Yet it does break another test we have which ensures that we don't use SoT values for fresh requests.
Almost, maybe we need a request type similar to StoreRequest.cached(key, refresh=true) but works the other way around where it allows falling back to SoT if fetcher throws an error or returns no results.
Something like: StoreRequest.fresh(key, fallackToSourceOfTruth=true).

Another issue we have is that, we don't have a construct that'll change the loading state of the Fetcher without it emitting any values. We don't really model loading state so maybe it is OK and it can be constructed from receiving disk values in a fresh request. Alternatively, we can consider adding StoreResponse.Empty but then we probably also want to emit it for SoT

[yigit]

actually that fallback should probably include cache as well, not just SoT

[yigit]

maybe we can do something like this:

fun <Key> fresh(key: Key, fallback:Fallback) 

Where fallback lets you specify a strategy. This case seems very similar to cases where people want to write get fresh data if it arrives within 300ms. It could be something like:

sealed class Fallback {
        class NoFallback internal constructor(): Fallback()
        class OnNoData() : Fallback()
        class WithTimeout(val timeoutInMs:Long) : Fallback()
        class WithDeferred(val lock: CompletableDeferred<Unit>) : Fallback()
        companion object {
            val NO_FALLBACK : Fallback = NoFallback()
        }
    }

and then you can call:

stream(StoreRequest.fresh(key, Fallback.OnNoData))

for this case or if you want timeout, you could do:

stream(StoreRequest.fresh(key, Fallback.WithTimeout(300)))

or we could also allow combining these

stream(StoreRequest.fresh(key, Fallback.WithTimeout(300).or(Fallback.OnNoData)))

We should also avoid complicating things at some point and ship a beta :D but this seems like a legit practical use case

[alexandercasal]

Would these solutions also apply to StoreRequest.cached?

From my example I'm using:

 myStore.stream(StoreRequest.cached(Key("value" ), refresh = true))

[yigit]

No this won't cover that actually 🤦‍♂️. We need a way to model Empty response for that case, which would require a new store response type.

[alexandercasal]

I'd need to think a little more about the StoreRequest.fresh scenario... But I wonder if the solution for StoreRequest.cached is simpler than creating a new StoreResponse type...

What if we changed the cache StoreRequest signature to something like:

fun <Key> cached(key: Key, shouldRefresh: suspend () -> Boolean) { ... }

This way the logic to check if the network call should even be executed is handled before the fetcher flow is started, preventing the Loading state from getting stuck, or even starting at all. I would expect under this implementation that the fetcher flow should always emit something...

Looking at it though, this change would affect fresh, cached, and skipMemory as the current refresh Boolean appears to be tied to the StoreRequest constructor...

[eyalgu]

I like modeling StoreResponse.NoNewData better than adding fallback functionality this close to beta. We can always add it in the next version:

For the cache case it will look like:
Cache
Source of truth
Fetcher - > Loading
Fetcher - > NoNewData

For the fresh case it will look like:
Fetcher - > Loading
Fetcher - > NoNewData
Cache
Source of truth

thoughts?