bugfix: reading from a silent stream should not fail with RedisCommandTimeoutException

[arturaz]

If the stream has no data for a long time, lettuce will throw RedisCommandTimeoutException.

In that case we don't want to fail the stream and we should restart the read.

Alternatively we should use the lettuce reactive API, but that requires quite a large of a refactoring.

[yisraelU]

TY

[yisraelU]

@arturaz I think I was a bit too hasty, I have had more time to look at this in detail.
Can you clarify in your scenario

• Whether the BLOCK option was used.
• If used, what timeout value was specified.

If block was not used , then a timeout exception occurred most likely due to underlying network latency issues (as you can see from the myriad of issues raised here

In that case perhaps this should be viewed as a feature and require opt in by the user

I would also expect in the test scenario added, where one explicitly sets a timeout in the client options, the stream would be shutdown in that case.

updated above

[arturaz]

Whether the BLOCK option was used.

My case uses default option, which is Some(Zero), meaning blocking indefinitely. But yeah, if there is no block, it should return immediately and suppressing the exception is bad.

I will take a look at rewriting this using the reactive redis API instead of async one to see if that helps.

[arturaz]

I migrated streaming to reactive Lettuce API, which, unfortunately, still raises the timeout exception if the stream is silent.

However, I still think it's a better option than repeatedly polling using async API. The simple use case (without restarts) is very straightforward now.

I also added restartOnTimeout: Option[FiniteDuration => Boolean] = None as an option to read.

[yisraelU]

Putting aside the switch to the Reactive API, as you are correct , as noted , it wont solve this problem.
The problem is inherent to how Lettuce works. Lettuce has 2 separate timeout configs .

• Connection timeouts
• Command timeouts

Using the default configs for a client would result in having a CommandTimeout set to 60 seconds as its inherited by default from the connection config which has a default of 60 seconds
The timeout options are set in Lettuce here

The command timeout mechanism is ignorant of the api choice (sync/async/reactive) and will timeout regardless .
One must either change the command timeout config duration, configure it to ignore certain commands, or not use a blocking command .

See this for discussion around CommandTimeout for blocking commands

In fact in the history of Redis4Cats , there is discussion if this was a wise decision to set block to 0 as default.

[arturaz]

So is there a problem with the added feature?

[yisraelU]

@arturaz apologies for the delay.
I am getting back to this now , and I think we should go back to your original commit , with handling/ignoring the RedisCommandTimeoutException .
This is effectively what is recommended by Lettuce for these scenarios (except they suggest the burden be handed to the caller and must define it in the TimeoutOptions )
By handling the CommandTimeout, we are providing a better UX to the user , and I believe what would be expected by a user when utilizing an fs2 Stream.

[arturaz]

I think we should go back to your original commit

Why? The reactive API is better suited to this scenario, rather than polling Redis via the async API and maintaining queues.

More over, I think the whole design of streaming API is broken due to use of the same F[_] for both streams and IO operations. See #962

[yisraelU]

I would like to dicsuss pros/cons of reactive an addition to the other issue you mentioned, I just think it should be a separate discussion and PR from this bug/ux fix, as it really is unrelated, and will be better served that way.

[arturaz]

OK, we can move reactivity to another PR, but can we please first discuss #963 as this is a dealbraker for otel4s-redis4cats integration?

[yisraelU]

yes, I'll give it a review. TY

[arturaz]

I removed the reactivity. Turns out, when converting push-based Publisher to pull-based Stream, you can't really do any buffering, so you have to have a chunk size of 1, which is ineffective, thus the current approach is better.

I left the (failed) attempt with the comment in https://github.com/arturaz/redis4cats/commit/484a28de18982d7396a80aafde91c8301d804f23

[yisraelU]

Looks good.
As this is breaking binary compatibility , lets create a new branch for the next major version. I am guessing the other PR is also breaking binary compatibility , so once all work is complete , we can merge and deploy new version

[arturaz]

Yes. In theory this could be not breaking bincompat, but that means adding an overload with no default values (due to Scala's limitation of two methods with default values unable to share the same name), which is... weird?

Can you review these PR's and give me LGTM, so I could prepare the joint version for you to review? On my local build there was a couple of minor merge conflicts.

[arturaz]

Also, what do you think of having a default Some(_ => true) for the reconnectOnTimeout?

[arturaz]

Argh... The same problem surfaces in subscribe and psubscribe. Need to add fixes there as well.

[arturaz]

That's weird. They kind of do not, but in my codebase I had this:

2025-02-05T18:12:38.424118962Z     io.lettuce.core.RedisCommandTimeoutException: Command timed out after 1 minute(s)
2025-02-05T18:12:38.424125403Z          at io.lettuce.core.internal.ExceptionFactory.createTimeoutException(ExceptionFactory.java:63)
2025-02-05T18:12:38.424131998Z          at io.lettuce.core.protocol.CommandExpiryWriter.lambda$null$0(CommandExpiryWriter.java:183)
2025-02-05T18:12:38.424154572Z          at io.netty.util.concurrent.PromiseTask.runTask(PromiseTask.java:96)
2025-02-05T18:12:38.424160815Z          at io.netty.util.concurrent.PromiseTask.run(PromiseTask.java:106)
2025-02-05T18:12:38.424166935Z          at io.netty.util.concurrent.AbstractEventExecutor.runTask(AbstractEventExecutor.java:173)
2025-02-05T18:12:38.424197963Z          at io.netty.util.concurrent.DefaultEventExecutor.run(DefaultEventExecutor.java:66)
2025-02-05T18:12:38.424205232Z          at io.netty.util.concurrent.SingleThreadEventExecutor$4.run(SingleThreadEventExecutor.java:997)
2025-02-05T18:12:38.424211358Z          at io.netty.util.internal.ThreadExecutorMap$2.run(ThreadExecutorMap.java:74)
2025-02-05T18:12:38.424217671Z          at io.netty.util.concurrent.FastThreadLocalRunnable.run(FastThreadLocalRunnable.java:30)
2025-02-05T18:12:38.424223891Z          at java.lang.Thread.run(Thread.java:1583)
2025-02-05T18:12:38.424230013Z          at delay @ dev.profunktor.redis4cats.effect.FutureLift$$anon$1.lift(FutureLift.scala:50)
2025-02-05T18:12:38.424236203Z          at fromCompletionStage @ dev.profunktor.redis4cats.effect.FutureLift$$anon$1.lift(FutureLift.scala:50)
2025-02-05T18:12:38.424242658Z          at fromCompletionStage @ dev.profunktor.redis4cats.effect.FutureLift$$anon$1.lift(FutureLift.scala:50)

The error message is kind of useless. Thus I have added .enhanceTimeoutException(s"xread(offsets=$offsets, block=$block, count=$count)") that injects the name into the exception, so it would be more useful.

[yisraelU]

Also, what do you think of having a default Some(_ => true) for the reconnectOnTimeout?

yes , I think we should . Otherwise it requires a user to be exposed to the inner workings of Lettuce and understand what CommandTimeout is.
I worry that could be confusing and perhaps mixed up with network timeouts.

[yisraelU]

I think this should be .always .
Keep in mind a RedisCommandTimeoutException is unrelated to network issues, rather its simpoly how long a command takes to run. When running a blocking command that waits for data, it will wait until there is data

[arturaz]

But I think it is related to network issues. If the network goes dark without any traffic on the line, you can hang there indefinitely, IIRC. Timeouts resolve that situation.

Maybe we should not provide a default here at all and force the user to choose an option.

[yisraelU]

There is another exception that is thrown for connection issues. This exception is not about network issues

[arturaz]

I have changed the default and removed enhanceTimeoutException, I think we can merge this now?

[yisraelU]

Go for it