[Bug] retry_policy not availably in activity info

[nathanielobrown]

What are you really trying to do?

From withing an activity I want to get the attempt number

Describe the bug

I pass in a retry_policy to workflow.execute_activity and expect to be able to get that policy from within the execute activity with activity.iinfo().retry_policy but I get None instead.

Minimal Reproduction

Not entirely self contained example, but I hope you get the idea.

I tried to make a failing test in this repo, but I'm still unable to run the development environment. I guess this is a separate issue, but running the tests fails with ImportError: dlopen(/Users/nob/repos/sdk-python/temporalio/bridge/temporal_sdk_bridge.abi3.so, 0x0002): tried: '/Users/nob/repos/sdk-python/temporalio/bridge/temporal_sdk_bridge.abi3.so' (mach-o file, but is an incompatible architecture (have 'x86_64', need 'arm64e')). My guess is that rust is cross compiling for x86_64 rather than the arm64e I need for my M1 Mac for some reason 🤷‍♂️.

from datetime import timedelta
import uuid
from temporalio import activity, workflow
from temporalio.client import Client
from temporalio.common import RetryPolicy
from bots_interface.temporal.server import temporal_worker_for_test


@activity.defn
async def return_retry_policy() -> RetryPolicy:
    activity_info = activity.info()
    # This is None!!!
    return activity_info.retry_policy


@workflow.defn
class Workflow:
    @workflow.run
    async def run(self) -> RetryPolicy:
        policy = await workflow.execute_activity(
            return_retry_policy,
            retry_policy=RetryPolicy(maximum_attempts=3),
            schedule_to_close_timeout=timedelta(seconds=5),
        )
        return policy


async def test_activity_info_retry_policy(temporal_client: Client):
    async with temporal_worker_for_test(
        temporal_client, Workflow, activities=[return_retry_policy]
    ) as worker:
        result = await temporal_client.execute_workflow(
            Workflow.run,
            id=str(uuid.uuid4()),
            task_queue=worker.task_queue,
            execution_timeout=timedelta(seconds=10),
            retry_policy=RetryPolicy(maximum_attempts=1),
        )
    assert result.maximum_attempts == 3  # This fails

Environment/Versions

• OS and processor: M1 Mac, MacOS 12.4
• Temporal Version: temporalio==0.1a2
• Are you using Docker or Kubernetes or building Temporal from source?: Using embedded golang temporal server as used in the repo. So compiling myself I guess

[cretz]

Thanks! Will investigate why retry policy is not there. Looks like I didn't properly assert for it at

sdk-python/tests/worker/test_activity.py

Line 78 in 0392468

async def test_activity_info(client: Client, worker: ExternalWorker):

. There may be a bug or two with what the server is returning on activity execution.

As for the cross-compile issue, you'll need to check rustup and Rust in general is defaulted to the arm target and not the x64 one. Once confirmed your default compilation target is arm, you can poe build-develop to get the right shared lib.

[nathanielobrown]

Ah, thank you. This explains it:

rustup show
Default host: x86_64-apple-darwin
rustup home:  /Users/nob/.rustup

stable-x86_64-apple-darwin (default)

I switched my default toolchain with rustup default stable-aarch64-apple-darwin, re-ran poe build-develop and now I can run the tests. Nice to learn a little something about rust.

[cretz]

I have confirmed via https://github.com/temporalio/sdk-core/blob/15ea5d3ee580e2a226eefdcf0d8444b2ae30a997/protos/local/temporal/sdk/core/activity_task/activity_task.proto#L55-L58 that we should not even expose this (that is a different kind of retry policy). None of our SDKs provide the original retry policy via the activity info (ref https://pkg.go.dev/go.temporal.io/sdk/internal#ActivityInfo, https://typescript.temporal.io/api/interfaces/activity.Info, https://www.javadoc.io/static/io.temporal/temporal-sdk/1.14.0/io/temporal/activity/ActivityInfo.html, etc) and Python shouldn't either.

[nathanielobrown]

Hmm, OK is there a different way for an activity to understand what retry policy it's operating under?

My use case is that my existing tasks (run using Dramatiq) log a Sentry errors and HTTP request logs on the last retry, which reduces logged errors to those that were not solved by retry. At some point I would also like to ask your opinion on the best way to do a Sentry integration (my guess is an interceptor?) but I don't want to ask too many questions in one day 🙂.

If there's a different approach you think could solve this problem, I would love to hear that as well.

[cretz]

Hmm, OK is there a different way for an activity to understand what retry policy it's operating under? [...] My use case is that my existing tasks (run using Dramatiq) log a Sentry errors and HTTP request logs on the last retry, which reduces logged errors to those that were not solved by retry.

The problem is that an activity may not even get called for its last attempt (what if it times out before execution?). However, if you still want to do something from the activity side on last attempt, you probably need to send that as a parameter to the activity, e.g.:

@dataclass
class MyActivityParams:
    some_other_param: str
    send_error_after_attempt: int

@activity.defn
async def my_activity(params: MyActivityParams):
    try:
        ...
    except Exception as err:
        if activity.info().attempt >= params.send_error_after_attempt:
            send_error(err)
        raise

This is untested of course, I just typed it here in issues.

But it can often be clearer and more accurate to react to activity failures when you get them back from the workflow instead of in the activity (which yes may mean starting another activity to send off the failure).

At some point I would also like to ask your opinion on the best way to do a Sentry integration (my guess is an interceptor?) but I don't want to ask too many questions in one day 🙂.

Ask away. Though I would recommend the forums at https://community.temporal.io/ or the #python-sdk Slack channel at https://temporal.io/slack for questions instead of GH issues.

[nathanielobrown]

The problem is that an activity may not even get called for its last attempt (what if it times out before execution?). However, if you still want to do something from the activity side on last attempt, you probably need to send that as a parameter to the activity, e.g.:

In practice, I do not find this to be an issue. For timeouts, all I really want to track is what percentage of the time a given task times out, I don't want Sentry errors because the details are not usually important and as a timeout can happen at any time you get a whole bunch of different error reports for one reason: the task is timing out.

The issue with logging activity failures from the workflow is I would lose all of the great context that Sentry collects, like local variables and breadcrumbs. For actually control flow due to the activity failures, I agree doing that in the workflow is best but I'm thinking more of observability of errors here.

It would be nice to have the retry policy because then you can have error handling contexts that can react to to the retry policy in a generic way without passing this for every activity. However, if it's not there it's not there and thank you for the example. At least I have the attempt number and I can make it work.

I'll do some experiments with Sentry setups and ask away when I'm a bit more oriented.

[cretz]

Closing. Feel free to reopen as necessary.