What's wrong with stream or how to use it properly?

[atonamy]

Let's review this simple example

  fun fetchFromCollection(keys: Set<MySerializedType>){
        someNonblockingScope.launch {
            for(key in keys) {
                myStore.steam(StoreRequest.cached(key, true)).collect {
                    if (it.origin == ResponseOrigin.Fetcher && it !is StoreResponse.Loading)
                        Log.d("CHECKING TRIGGER", "$key triggered")
                }
            }
        }
    }

the expected outcome should be that CHECKING TRIGGER will trigger once for each unique key even if fetchFromCollection executed several times (each time always unique keys)?

If I execute fetchFromCollection with one unique set of keys it will run as expected, but if I execute second time fetchFromCollection with another different set of keys it will trigger CHECKING TRIGGER condition more than once(new set of unique keys).

What I miss? Why it doesn't work as expected?

And then if execute fetchFromCollection three, four times and so on each time with new set of keys stream will just hang with Loading state forever.

I pushed this example project demonstrating the issue in full scale.

[eyalgu]

Hello, Can you please help by providing a minimal repro. This includes:

1. The store creation code you are using
2. The actual key types, which for simplicity sake should be a basic type like String (to rule out bugs resulting from issues with hashCode impl)

Thanks for the report!

[atonamy]

1. For the key I just use java BigInteger but I believe u can reproduce with String as well.

private val blockInfoStore = StoreBuilder
        .fromNonFlow<BigInteger, GetBlockResponse> {
            rpc.getBlock(GetBlockRequest(it.toString()))
        }.cachePolicy(
            MemoryPolicy.builder()
                .setMemorySize(5)
                .setExpireAfterAccess(10)
                .setExpireAfterTimeUnit(TimeUnit.MINUTES)
                .build()
        )
        .build()

if u use get or fresh extension functions no issue at all. The problem only with stream itself.

You can clone code I left link above and test yourself.

[eyalgu]

On first glance - you're not providing a local source of Truth (disk cache) and you are setting your memory cache to expire. I this case the fetcher will be called when the cache is evicted. Can you confirm by one of the following:

1. removing the cache expiration
2. add a source of Truth

[atonamy]

local source of Truth is database or what? Removing the cache expiration still trigger the several times with same state and origin and key.

I even tried very simple configuration:

StoreBuilder
        .fromNonFlow<BigInteger, GetBlockResponse> {
            rpc.getBlock(GetBlockRequest(it.toString()))
        }
        .build()

Still similar result. Please try my example I submitted here early.
How to add disk cache as simple file without setup database?

Updated:
But at least if I remove cache expiration stream never hang with Loading state.

[yigit]

are you using alpha01? we had a bug that always triggered fetcher when there is no source of truth, we've fixed it in alpha02.
#75

[eyalgu]

For file based persister, you can check out com.dropbox.mobile.store:filesystem:4.0.0-alpha02 It's definitally not as well supported as store but might be good for your needs.

[atonamy]

@eyalgu

Failed to resolve: com.dropbox.mobile.store:filesystem:4.0.0-alpha02

@yigit
Chek source code in example is using alpha02.

You know even if use StoreRequest.fresh the behaviour still the same.

Then if persister is so essential and mandatory what is point to use store if I only need temporary memory/file cache for example? At least library should not allow to build store without persister and throw error during compilation (or at runtime).

[ychescale9]

Correct coordinate is com.dropbox.mobile.store:filesystem4:4.0.0-alpha02.

[eyalgu]

Hi @atonamy, in order for allow us to investigate. Can you please provide a minimal repro. The current repro you provided includes rpc.getBlock etc. that makes it impossible for us to repro. Can you provide something very basic e.g a fetcher that translates an Int to it's String. (Extra cupcake points if you can just put it in a self contained junit test)

[yigit]

I've forked your app and added more logging,
it is always getting new ids from block numbers.

If you look at these logs:

yigit/store4-issue@02e09bd

You can see that it always gets a new block id so it is a new key for a new request.
Btw, it did help find a bug in cache though 👍
#89

I'm not sure if i'm looking at the right part of the code, as @eyalgu mentioned, it would be much more clear to understand if this was a sample app but from what i can see in this app, it keeps sending different keys all the time hence getting new values.

here is a sample output from a run:

2020-02-03 21:22:18.777 13201-13252/? I/System.out: request for 103388547. seen before? false, []
2020-02-03 21:22:19.036 13201-13241/? I/System.out: request for 103388546. seen before? false, [103388547]
2020-02-03 21:22:19.292 13201-13242/? I/System.out: request for 103388545. seen before? false, [103388547, 103388546]
2020-02-03 21:22:19.540 13201-13255/? I/System.out: request for 103388544. seen before? false, [103388547, 103388546, 103388545]
2020-02-03 21:22:19.792 13201-13252/? I/System.out: request for 103388543. seen before? false, [103388547, 103388546, 103388545, 103388544]
2020-02-03 21:22:38.410 13201-13255/? I/System.out: request for 103388584. seen before? false, [103388547, 103388546, 103388545, 103388544, 103388543]
2020-02-03 21:22:38.663 13201-13252/? I/System.out: request for 103388583. seen before? false, [103388547, 103388546, 103388545, 103388544, 103388543, 103388584]
2020-02-03 21:22:38.913 13201-13241/? I/System.out: request for 103388582. seen before? false, [103388547, 103388546, 103388545, 103388544, 103388543, 103388584, 103388583]
2020-02-03 21:22:39.166 13201-13255/? I/System.out: request for 103388581. seen before? false, [103388547, 103388546, 103388545, 103388544, 103388543, 103388584, 103388583, 103388582]
2020-02-03 21:22:39.422 13201-13240/? I/System.out: request for 103388580. seen before? false, [103388547, 103388546, 103388545, 103388544, 103388543, 103388584, 103388583, 103388582, 103388581]

Lmk if i'm looking at the wrong thing.

[atonamy]

@yigit

Yes this #89 crashing issue also exists but if you have a look log carefully you may notice with this different keys Flow.collect {} triggers with the same origin and status more than one time (but must be one only right?) and keep increasing this duplicate calls when more different keys you fetch which is obviously somewhere issue there.

Sorry I'm a bit busy, but I can create more testable solution maybe on weekends if it still valuable.

[eyalgu]

Not sure if relevant, but in your original example I noticed that you launch one coroutine and in it iterate over all your keys and collect the streams on them. A store stream does not finish collecting, so in that example the code will only start to collect the first key and will never get to the second.

Again that may be just specific to the example you have so a full repro would be the best next step here.

Thanks again sure trying out store and taking the time to report bugs!

[atonamy]

@eyalgu but collection streams always happening with new unique keys in this example. So even if previous streams is not finished how come it affect new streams which is not related to previous? And u can even wait when previous stream collection will finish you still will get this duplicate calls on collection with new stream and with new different keys anyway. You can even try to cancel coroutine after complition which collect stream and new stream still produce duplicate collection with same origin and state.

Update:
I suspect the main issue when we collect streams in parallel, if we do it sequently strictly one by one the store with streams works fine.

Need more tests to confirm.

[yigit]

Flow.collect {} triggers with the same origin and status more than one time

In my log, it always gets called w/ a new key so store makes 1 request for each new key. Not sure if I understand the problem correctly, that code seems to work as expected. If there are different keys, there will be different requests. I feel like i'm still not understanding what the problem is :/.

Every time you call the runStoreStreamTest function, you are asking for a fresh value so it skips cache, calls rpc.info.headBlockNum which returns new keys (from its previous call) and the query to the second store always receives new keys, hence makes new requests.

[atonamy]

When we collect stream with cache and new key it first goes to fetcher since we don't have a cache yet because of new key so the first we receive in collect loading status and fetcher origin right? Then once data received successfully from remoute source collect will call one more time and show in the status that we have data now correct?

Now if start to collect multiple streams in parallel the above described procedure the same for each stream with new unique key ?

Then if we split multiple streams collection in several batches with equal quantity of parallel streams which is run sequentially when one batch of multiple streams collection is complete and start another one we can see that described above behavior is different now the successful status with fetcher in collect called twice now in our second batch which is shouldn't be like this? And if we continue to run third, fourth batches sequentially we can see that collect calls keep increasing with same status and origin like in this described case.

Do I clearly described the problem?

In all streams and batches the keys is always new and unique never reuse same key anywhere in any stream in this scenario.

[digitalbuddha]

You had me until the last paragraph. To be honest the simplest way to get a fix is to submit a failing test case. This will both allow us to reproduce your issue and verify a fix. Seems like you might be hitting on some use case we did not consider.

[atonamy]

@digitalbuddha no problem let's wait untill end of the week I'll try my best to write unit tests on weekend with isolated and mocking infrastructure if it still will be the case.

thank you

[digitalbuddha]

Closing, please reopen when you have a verifiable test case