Fixes an issue that happens with OpenTelemetryMiddleware when a message is requeued with Context.requeue().
Context.requeue() uses self.broker.kick to kick a message to the broker which skips the Middleware pre-send and post-send invocations that kicker does here:
Messages that are requeued with the same context variables will raise an error on post_save's .detach() because they are now running in a different async context.
The message's lifecycle on OpenTelemetryMiddleware will be:
pre_send -> message.labels are injected with context here
post_send
pre_execute -> message.labels are extracted here and re-used
post_execute
requeue() happens -> skipping pre_send and post_send, therefore not renewing the context.
pre_execute -> context inferred from message.labels again
post_execute
post_save -> Detaching the stale context here. Raises ValueError: Token was created in a different context
This change gets rid of post_save on OpenTelemetryMiddleware to detach the context on post_execute instead. Since, we are not actually doing anything database save related on the current post_save method, it seems fair to move everything under post_execute as this is where the task execution actually ends.
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
Fixes an issue that happens with OpenTelemetryMiddleware when a message is requeued with Context.requeue().
Context.requeue() uses self.broker.kick to kick a message to the broker which skips the Middleware pre-send and post-send invocations that kicker does here:
taskiq/taskiq/kicker.py
Lines 160 to 162 in ced1909
taskiq/taskiq/kicker.py
Lines 168 to 170 in ced1909
Messages that are requeued with the same context variables will raise an error on post_save's .detach() because they are now running in a different async context.
The message's lifecycle on OpenTelemetryMiddleware will be:
This change gets rid of post_save on OpenTelemetryMiddleware to detach the context on post_execute instead. Since, we are not actually doing anything database save related on the current post_save method, it seems fair to move everything under post_execute as this is where the task execution actually ends.