"Redis query EventsByTag must not see new events after complete" test is not implemented properly about akka-persistence-redis

[satabin]

Hi,

With the current implementation I would say that the tested behavior is the implemented one. The semantics of currentEventsByTag in the current implementation is a bit different from the one of LevelDB backend. Events occurring after source opening may be seen by the source due to how event retrieval is implemented. The implementation buffers events and closes the source when no events are found in the database when refilling buffer.

I agree this brings less guarantees to the user and may lead to apparently strange behaviors.

It should be possible to implement the same semantic than the LevelDB backend, and I will investigate how this can be achieved. In any way, documentation should be more explicit about the behavior.

Thanks for reporting.

[alexvaluyskiy]

All known implementations of Current* queries (LevelDb, Cassandra, Jdbc) fetch a data from a storage only once. Otherwise, it would be something like live query

[satabin]

Not necessarily, you can retrieve data in batch up to a certain point in time (say request time). You can’t always (or even wish to) load all events into memory at once. E.g. you might want to keep the memory footprint as low as possible.

[satabin]

I investigated and there is an easy way to implement the same semantics as for LevelDB, I will work on it ASAP.

[alexvaluyskiy]

I've found the same problem for CurrentEventsByPersistenceId query

[satabin]

Yes, I implemented all the Current* streams the same way and will change their semantics.

[alexvaluyskiy]

I see the similar issue in return remaining values after partial journal cleanup test. You have requested two items but should have requested only one item. The current stream should be completed right after delivering the whole buffer. Maybe you should change this line https://github.com/safety-data/akka-persistence-redis/blob/master/src/main/scala/akka/persistence/query/journal/redis/EventsByPersistenceIdSource.scala#L213 to if (buffer.isEmpty && (currentSequenceNr > to || !live))

[satabin]

Yes it seems this condition is missing. I will perform a rewriting and complete review of implementation and tests.