Allow for pipelines to ignore error and continue.

[J08nY]

It is useful for me to have a pipeline of messages that execute. However, I want to keep executing the pipeline even if the individual message fails. This pull-request implements that using a new option on the message in the Pipelines middleware.

[synweap15]

@J08nY I'd rename the left-side to ignore_exception

[synweap15]

@J08nY exception is not None and message.failed shouldn't be considered equal - they may have different meanings in the actor implementation. Only explicitly thrown exceptions should be caught.

I'd also go for explicit .. and ignore_exception is not True

[J08nY]

That is why I had the name "errorok", to consider both cases of message failures. This can of course be split, but that requires a new parameter.

[synweap15]

@J08nY with the above, the changes would require a redo

[LincolnPuzey]

I believe we are trying to avoid adding new features when possible. Given @J08nY it looks like you could achieve what you needed with a middleware subclass, I am apprehensive about bringing the change into the core library, sorry.

---

Looking at the code, if we allow pipelines to continue when exception is not None (but the message hasn't failed yet), I think that could mean the Message will be re-tried and the pipeline is continued.

Which could then mean the pipeline would be continued multiple times? each time the message is re-tried (default 20 times). This is obviously not good.

[J08nY]

I believe we are trying to avoid adding new features when possible. Given @J08nY it looks like you could achieve what you needed with a middleware subclass, I am apprehensive about bringing the change into the core library, sorry.

I see. Would you mind including it somewhere in the README? I mean like a recommendation to try to use middleware for custom features, or make an issue if it is not possible and that new features are unlikely to get accepted. Would save time.

Looking at the code, if we allow pipelines to continue when exception is not None (but the message hasn't failed yet), I think that could mean the Message will be re-tried and the pipeline is continued.

Hmm, I guess my understanding of what the different states can be and how they are handled is lacking.

Would you suggest subclassing the Pipeline middleware and replacing the built-in one? In my use-cases I do not have/want retries and thus it is much simpler.

[synweap15]

@LincolnPuzey @Bogdanp gave me the go-ahead. Let's have a private conversation.

@J08nY, I apologize for the poor communication.

[LincolnPuzey]

@synweap15 My apologies, I didn't know.

Do you agree that continuing a pipeline multiple times due to retries, is something we should avoid?

[J08nY]

Are there some semantics of this concept that you want to ignore exceptions in an actor in a pipeline, that work together with retries?

The only thing I can think of is to make the actor eat the exception and thus effectively disable retries.

[J08nY]

Any updates on this? I would still like to be able to have pipelines with actors that fail, yet the pipeline continues.

[synweap15]

Hey, it's on my radar, I'll get back to it soon. wt., 10 cze 2025 o 13:09 Ján Jančár ***@***.***> napisał(a):

…

*J08nY* left a comment (Bogdanp/dramatiq#679) <#679 (comment)> Any updates on this? I would still like to be able to have pipelines with actors that fail, yet the pipeline continues. — Reply to this email directly, view it on GitHub <#679 (comment)>, or unsubscribe <https://github.com/notifications/unsubscribe-auth/AAWT3D3XXU4DOQGOHR7BJQ33C24GLAVCNFSM6AAAAABWO72LGWVHI2DSMVQWIX3LMV43OSLTON2WKQ3PNVWWK3TUHMZDSNJYG42DCNRXHA> . You are receiving this because you were mentioned.Message ID: ***@***.***>

[synweap15]

I see this feature useful in scenarios where an action may fail, but there are several attempts to execute it. Example I have encountered - heavy calculations result caching where the cache key may be locked. Not a deal breaker if fails but desirable to be persisted.

Explaination of params use

• pipe_ignore=[any], retries=10, pipe_ignore_exception=False -> action fail - retries actor, halts on final failure (as it is now)
• pipe_ignore=[any], retries=10, pipe_ignore_exception=True -> action fail - retries actor, then continues pipeline on fail
• pipe_ignore=True, retries=[any], pipe_ignore_exception=True -> action fail - next actor runs with no input from prev
• pipe_ignore=False, retries=[any], pipe_ignore_exception=True -> action fail - next actor runs and either the output value or None is passed (more on that below)

I guess this would be similar in function to a middleware:

class GracefulRetryMiddleware(Middleware):
    def after_process_message(self, broker, message, actor, result=None, exception=None):
        if exception is not None:
            retries_left = message.options.get("retries", 0)
            max_retries = actor.options.get("max_retries", 0)
            
            if retries_left >= max_retries:
                # Final retry failed, continuing anyway
                return None  # swallow the exception after final retry
            
            # "Exception: {exception}, will retry (attempt {retries_left}/{max_retries})"
            raise exception 

Considerations

• we lose information whether or not the previous actor ran successfully.
• if pipe_ignore=False and pipe_ignore_exception=True, the input to the following actor may be None - is None check may be needed

Thoughts?