Hey there! I’m trying to implement some sort of rudimentary Sensor (like given here https://github.com/PrefectHQ/prefect/discussions/4156), but I would really like to implement a timeout that spans retries. Or in other words, enforce max_retries even though I am raising the RETRY signal manually (which ignores max_retries). Any ideas how to tackle this? I could store a counter in the context and increment a counter inside each of my task’s run() calls I suppose

I am doing something similar, I wanted max retries based on a dynamic variable in my DB. I found you can get the retry number like this:

retry_count = prefect.context.get("task_run_count", 1)

Robert's solution sounds excellent for your use case. If you do need to persist information other than

task_run_count

, you could use KV store https://docs.prefect.io/orchestration/concepts/kv_store.html

I have seen this done using an infinite loop or recursion in the task to poll until a condition is met.