@soyuka This isn't necessarily directed at you, but I think just re-queing failed items or items that have been left in processing too long isn't a very safe way to go about this.

In my experience, when a message/task fails, it almost always fails again when re-run, and you end up just having 6 failed attempts instead of the one. And in some cases, if your system is connecting to external systems, it could end up causing unwanted duplicate data.

Not saying that retrying isn't a useful feature, but I think it should be something that should at least have the option to be turned off or maybe put into a failed jobs log or table.

https://github.com/krakphp/symfony-messenger-redis Here is the redis bundle I made and I admittedly copied several aspects of the redis part between thehttps://github.com/krakphp/job and your bundle.

but I think it should be something that should at least have the option to be turned off or maybe put into a failed jobs log or table.

Definitely!

I see that in your bundle you're using one queue per connection, therefore you are storing different kinds of messages on the same queue. Isn't this an issue while debugging or maintaining these lists? Do you have an opinion about this?
Thanks for the early comments!

@soyuka Well no because it's more like one queue per transport.

framework:  messenger:    transports:      main:        dsn: '%env(TRANSPORT_DSN)%'        options:          queue: queue_main      secondary:        dsn: '%env(TRANSPORT_DSN)%'        options:          queue: queue_secondary    routing:      App\MainMessage: main      App\SecondaryMessage: secondary

That would be an example configuration. This definitely allows multiple types of messages for a single queue/transport, but that just depends on how you want to scale out your transports. Some not so frequent messages can go on the main queue, but maybe a heavy traffic message might go on it's own so it can have a dedicated consumer and not get blocked.

@soyuka This isn't necessarily directed at you, but I think just re-queing failed items or items that have been left in processing too long isn't a very safe way to go about this.

I've decided to follow@sroze implementation for Amqp which states that:

If something goes wrong while consuming and handling a message from the Redis broker, there are two choices: rejecting or re-queuing the message.
If the exception that is thrown by the bus while dispatching the message implements this interface (RejectMessageExceptionInterface), the message will be rejected. Otherwise, it will be re-queued.

@soyuka Extension looks great! Thanks for the hard work!

@soyuka in regards to Requeing the message by default and Rejecting if it implements a RejectMessageExceptionInterface

I think it would make more sense to default to Reject, and requeue if they implement a RequeueMessageExceptionInterface; however, that's my opinion for reasons as stated previously in this thread. If we end up sticking with defaulting to Requeue, do you think it would make sense to add a middleware/option in the messenger config to basically wrap all exceptions into a RejectMessageExceptionInterface?

do you think it would make sense to add a middleware/option in the messenger config to basically wrap all exceptions into a RejectMessageExceptionInterface?

Looks a bit too magic to me. I definitely get your concerns about requeuing though, and I'd be in favor of having aRequeueMessageExceptionInterface, with the default behavior set to rejecting messages that fail.

@soyuka with the release of redis streams in redis 5.0, i'm wondering if we should target that data structure for a more robust implementation. We also could possibly implement two different redis connections one for streams and one with lists.

Ping@soyuka! If you're still willing to work in this, we just merged#30557, which clarifies what a transport needs to do / behave (should simplify the receiver a bit). I would love to see this PR updated!

Will definitely work on this thanks for the heads up, any idea if the related documentation is going to be updated?

/edit: I've looked at the PR, messenger is improving that's nice! I've an issue with the new interface though, because my Connection doesn't uses a Enveloppe for now (inreceive I'm working on a message, not anEnvelope).

@soyuka Awesome!

I've looked at the PR, messenger is improving that's nice! I've an issue with the new interface though, because my Connection doesn't uses a Enveloppe for now (in receive I'm working on a message, not an Envelope).

Can you tell me more about this? Or you can ping me on Slack. Do you mean the methods likeReceiverInterface::ack(Envelope $envelope)? If that's true, it shouldn't be a problem because your receiver is not responsible for calling this anymore - theWorker will call this and it will pass you the Envelope. If you need some sort of "identifier" so you know which message to ack (you will), you should add a stamp to the Envelope before passing it to$handler. Check out the AmqpReceiver. We wrap in a custom stamp (https://github.com/symfony/symfony/blob/master/src/Symfony/Component/Messenger/Transport/AmqpExt/AmqpReceiver.php#L73) then use that inack() andreject() to know which message was received, so we can ack/reject it.

groups are awesome if you have multiple systems receiving the same messages

I have no idea what these streams are all about, what is the awesomeness of them in this use-case? When you say that "most things is handled by redis itself" that sounds appealing... but what do you mean exactly? What are the use-cases/logic we wouldn't have to implement anymore?

the only con is that it only works with redis 5 and ext-redis 4.2.

ext-redis 4.2 has been released on the 2018-11-20 so I don't see this as an issue. The 4.3 has also been released and is considered as stable. Though, do you know what's the Redis 5 availability in most of the cloud providers?

I don't think that streams are complex in case of the php implementation

I just looked at theredis introduction and it seems to have more features compared to using a normal list. This was the complexity I was talking about, not the one from the php implementation.

But I'm nobody so I can't do a decision here, just wanted to mention it here, thought the implemtation could maybe help.

Everyone's thought is appreciated! I wasn't even aware that redis had implemented this feature, so thanks for the discovery and I'll definitely take a look at your implementation!

My only concern with using streams (other than the additional complexity), is that forcing the use of the ext-redis extension can be an issue under heavier workloads.

We were getting intermittent errors and failures on redis consumer with sf messenger and after some investigation, it looks like they are just related to the redis-ext and those errors don't occur on predis.

phpredis/phpredis#831
predis/predis#524 (comment)

After we switched, those intermittent errors all went away..., However, predis doesn't seem actively maintained, It's a good library and seems to be more reliable than phpredis, but nrk/predis hasn't touched that repo in forever. Makes me think that a reliable fork or a new maintainer might be necessary for that library.

@ragboyjr if we want this to get merged into symfony we have to use the officialext-redis, we can't introduce a dependency to a third party because of maintenance issues.

I have no idea what these streams are all about, what is the awesomeness of them in this use-case? When you say that "most things is handled by redis itself" that sounds appealing... but what do you mean exactly? What are the use-cases/logic we wouldn't have to implement anymore?

Lets say you have 3 Systems. System A is sending messages and System B and C want to receive this messages. When you want that B and C receive both all messages without redis streams you need to implement that logic yourself or sending them into 2 lists or something (correct me if I'm wrong). With redis streams you can just create 2 groups. one for system b and one for system c and redis will give you the messages which you did not receive yet.

I personally found it really simple adding a new messages to a stream can be done with:

$this->redis->xAdd('STREAMNAME','*', ['content' =>json_encode($encodedMessage)]);

Reading the message can be done with:

$messages =$this->redis->xReadGroup('GROUPB','CONSUMER1', [$this->stream =>0],1,45);

Redis automatically sets the read messages into a pending state for this group so if you example have for the same group multiple consumer they will not get the same message again.

And after you did successfully process the message you mark it in your group as acked:

$this->redis->xAck('STREAMNAME','GROUPB', [$messageId]);

So if you later need a System D also receiving all messages you just create in redis a GroupD and so system D is also starting to receive all messages.

So streams will shine if you have multiple systems receiving messages. If you just have 2 Systems a Sender and a Receiver there will be no differents. But if you have multiple receivers Redis will handle you which group/system did receive a message and which one doesn't.

The following is also a interesting blog post about streams:https://brandur.org/redis-streams.

it seems to have more features compared to using a normal list. This was the complexity I was talking about, not the one from the php implementation.

I think as the streams feature is more build for this usecase less complex because of its features and automatic mechanisms.

But maybe I have to less knowledge about the current implementation and maybe it makes sense to have 2 redis transports one using the redis list's and one using redis streams.

Indeed, if there is this logic of parking the messages until they are acknowledged, this sounds very interesting because the less we have to deal with these distributed system issues, the better 🙃

@alexander-schranz many thanks about this detailed explanation, it helped me to understand better why redis streams may be good for this use case! I'll definitely look into that and use them in this transport implementation!

@soyuka I think the SF Lock and SF Cache component are able to utilize both redis libraries, i'd imagine if we did a redis transport using the list strategy, we should be able to provide some sort of compat layer to using either of the redis libraries.

I'd be willing to help if needed, but as someone who has used a redis transport with messenger at a high volume on both ext-redis and predis, I wouldn't feel comfortable only allowing ext-redis.

Ping! Do we have some direction/motivation/time to finish this? We're after feature freeze, though there is some wiggle room probably because this component is experimental.

Ping! Do we have some direction/motivation/time to finish this? We're after feature freeze, though there is some wiggle room probably because this component is experimental.

To me, time is an issue, maybe this week-end but definitely not sooner. Also, the streams must be tested they may improve the transport a lot, which means more work.

To me, time is an issue

I hear that :). Good candidate maybe for the hackathon this weekend, if not by you - by someone else? We could check to see if anyone has the expertise & is interested.

Cheers!

I'd love to work on this this week end but as I'm the Api Platform referent I'm not sure that I'll be able too :p.

@weaverryan@soyuka I will have a look at the current state at theEU-FOSSA Hackathon and will then refractor it to use the redis stream functions.

Added the hackathon star - thank you@alexander-schranz!

Analysed the current state and after also having a look at the new ReceiverInterface and playing around about it my concept will be the following:

+-----------R|    GET    | -> XREADGROUP+-----------+      |      | handleMessage      V+-----------+  No|  failed?  |---------------------------++-----------+                           |      |                                 |      | Yes                             |      V                                 |+-----------+  No                       ||   retry?  |---------------------------++-----------+                           |      |                                 |      | Yes                             |      V                                 V+-----------R                    +-----------R|   REJECT  | -> XACK            |    ACK    | -> XACK+-----------+                    +-----------+

GET: Will useXREADGROUP to read the one message from the stream
REJECT: Reject will just remove the message withXACK from the stream as adding it back to the stream is handled by symfony worker itself
ACK: Will use theXACK Method to ack the message for the specific group

The sender will still be simple by calling theXADD redis function.

#EU-FOSSA

I don't know about the Redis streams part, but I can confirm the rest of the diagram is perfect: you're acking/nacking with the correct logic.

@alexander-schranz great. Fancy updating this pull-request (or creating another one)?

@sroze If I can I will update this PR so we have all discussion about it here.

Please feel free to update it!

I'm not able to push into@soyuka's fork so I created a new PR#30917
/cc@Nyholm@sroze

Closing this one then!