9 - 错误处理

Temporal 通过重试和持久化执行自动处理多种类型的失败。本节展示如何在此基础上构建健壮的错误处理。

关键概念：

并非所有失败都应一视同仁地处理。瞬时失败（比如短暂的网络抖动）通常自行恢复，应立即重试。间歇性失败（比如速率限制）需要递增重试延迟。永久性失败（比如无效输入）无法通过重试解决，需要修改数据或代码。

Temporal 区分工作流任务失败（可通过重新部署修复的 Bug）和工作流执行失败（应停止工作流的业务逻辑失败）。任务失败会自动重试，因此可以在不丢失状态的情况下修复并重新部署。执行故障需要开发者显式抛出 ApplicationError。

使 Activity 具备幂等性

如何使用 Temporal Python SDK 使 Activity 具备幂等性

由于 Activity 可能因失败而重试，强烈建议使其具备幂等性。幂等操作无论执行一次还是多次，都产生相同结果。

Activity 遵循至少一次（at-least-once）执行模型。如果 Worker 成功执行 Activity，但在通知 Temporal 服务端之前崩溃，则该 Activity 将被重试。如果 Activity 不具备幂等性，可能导致在支付处理中重复扣款，或在基础设施配置中重复创建资源。

使用幂等键

大多数外部服务都支持幂等键——防止重复操作的唯一标识符。当服务端收到带有已处理过的键的请求时，直接返回原始结果，而不再次执行该操作。

通过组合工作流运行 ID（Workflow Run ID）和 Activity ID 创建幂等键：

from temporalio import activity

@activity.defn
async def process_payment(amount: float, account: str):
    info = activity.info()
    idempotency_key =f"{info.workflow_run_id}-{info.activity_id}"

    # Pass idempotency_key to your payment service
    result = await payment_service.charge(
        amount=amount,
        account=account,
        idempotency_key=idempotency_key
    )
    return result

该值在 Activity 重试期间保持不变，在所有工作流执行之间是唯一的。

将 Activity 设计为原子操作

Activity 是原子性的——要么成功完成，要么不完成。如果 Activity 执行多个步骤，最后一步失败，那么重试整个 Activity。

以这个 Activity 为例：

在数据库中查找数据

使用该数据调用微服务

将结果写到文件系统

如果第 3 步失败，重试时将再次执行全部三个步骤。可以将此拆分为三个独立的 Activity，以便仅重试失败的步骤。但需要权衡，这样做导致事件历史更大，包含更多的 Activity 执行。

从 Activity 中抛出异常

如何使用 Temporal Python SDK 从 Activity 中抛出异常

使用 ApplicationError 传递来自 Activity 的应用程序特定失败。显式地抛出该异常，使开发者获得更强的控制。

在 Temporal Activity 中，从 Python 代码抛出的其他异常都会被转换为 ApplicationError。这样，错误的类型、严重程度以及任何附加详细信息都可以发送到 Temporal 服务，在 Web UI 中索引，甚至可以跨语言边界序列化。

from temporalio import activity
from temporalio.exceptions import ApplicationError

@activity.defn
async def validate_charge(credit_card_number: str, amount: float):
    if not is_valid_card(credit_card_number):
        raise ApplicationError(
            f"Invalid credit card number: {credit_card_number}",
            type="InvalidCreditCard",
        )

    if amount <= 0:
        raise ApplicationError(
            f"Amount must be positive, got {amount}",
            type="InvalidAmount",
        )

    return True

当抛出 ApplicationError 时：

提供描述性 message

可选地提供用于对失败进行分类的 type 字符串

错误以 ActivityTaskFailed 事件的形式出现在事件历史中

当 Activity 失败时，Temporal 将异常包装在 ActivityError 中，然后再将其传递给工作流。ActivityError 提供的上下文包括：

失败的 Activity 类型

重试次数

原始原因（开发者抛出的 ApplicationError，或 TimeoutError、CancelledError 等）

从工作流中抛出异常

如何使用 Temporal Python SDK 从工作流中抛出异常

其行为取决于抛出的是哪种异常：

使工作流执行失败

要使工作流执行失败，可抛出 ApplicationError：

from temporalio import workflow
from temporalio.exceptions import ApplicationError

@workflow.defn
class PizzaDeliveryWorkflow:
    @workflow.run
    async def run(self, order):
        distance = await workflow.execute_activity(
            calculate_distance,
            order.address,
            start_to_close_timeout=timedelta(seconds=10)
        )

        if order.is_delivery and distance.kilometers > 25:
            workflow.logger.error("Customer outside service area")
            raise ApplicationError(
                "Customer lives outside the service area",
                type="CustomerOutsideServiceArea"
            )

        # Continue with order...

使用 Temporal ApplicationError 类的原因之一是，它支持设置 non_retryable 参数。通过该参数，开发者可以决定是否不自动重试。这在因输入数据错误而故意使工作流失败时非常有用。也可以在 Activity 重试策略中指定不可重试的错误列表。

这使工作流执行进入“Failed”状态，且不会自动重试。可将其用于重试无济于事的永久性失败——比如客户距离过远。

触发工作流任务重试

抛出其他 Python 异常（比如 ValueError 或 TypeError）将导致工作流任务失败，该任务将自动重试：

# This causes a Workflow Task failure (retries automatically)
raise ValueError("Unexpected condition")

这是有意为之的设计。普通 Python 异常被视为可通过代码部署修复的 Bug，而非业务逻辑失败。工作流任务无限重试，可以在不丢失工作流状态的情况下修复 Bug，重新部署。

在工作流中处理异常

如何使用 Temporal Python SDK 在工作流中处理异常

在工作流中使用 Python 的 try/except 语句块处理 Activity 失败：

from temporalio import workflow
from temporalio.exceptions import ActivityError, ApplicationError
from datetime import timedelta

@workflow.defn
class MoneyTransferWorkflow:
    @workflow.run
    async def run(self, details):
        # Withdraw money
        try:
            withdraw_result = await workflow.execute_activity(
                withdraw,
                details,
                start_to_close_timeout=timedelta(seconds=10)
            )
        except ActivityError as e:
            raise ApplicationError(
                f"Withdrawal failed: {e.cause}",
                type="WithdrawalError"
            )

        # Deposit money
        try:
            deposit_result = await workflow.execute_activity(
                deposit,
                details,
                start_to_close_timeout=timedelta(seconds=10)
            )
        except ActivityError as e:
            # Deposit failed - attempt refund
            try:
                await workflow.execute_activity(
                    refund,
                    withdraw_result,
                    start_to_close_timeout=timedelta(seconds=10)
                )
                raise ApplicationError(
                    f"Deposit failed but money refunded to source account",
                    type="DepositError"
                )
            except ActivityError as refund_err:
                raise ApplicationError(
                    f"Deposit failed and refund also failed: {refund_err.cause}",
                    type="CriticalTransferError"
                )

        return f"Transfer complete: {withdraw_result}, {deposit_result}"

可以在工作流中捕获的常见 Temporal 异常：

ActivityError —— Activity 在耗尽重试次数后失败

ChildWorkflowError —— 子工作流失败

CancelledError —— 工作流、Activity 或定时器被取消

TimeoutError —— 操作超时

如果这些异常未被捕获而向上传播，工作流执行将失败（对于 CancelledError，则进入“Canceled”状态）。

配置自定义重试策略

如何使用 Temporal Python SDK 配置自定义重试策略

Activity 默认使用无限重试和指数退避的重试策略。可以根据预期的失败模式对此进行自定义。

from temporalio import workflow
from temporalio.common import RetryPolicy
from datetime import timedelta

@workflow.defn
class OrderWorkflow:
    @workflow.run
    async def run(self, order):
# Custom retry for rate-limited service
        retry_policy = RetryPolicy(
            initial_interval=timedelta(seconds=10),
            backoff_coefficient=3.0,
            maximum_interval=timedelta(minutes=5),
            maximum_attempts=20,
        )

        result = await workflow.execute_activity(
            call_external_service,
            order,
            start_to_close_timeout=timedelta(seconds=30),
            retry_policy=retry_policy,
        )
        return result

重试策略属性：

initial_interval：首次重试前的延迟时间（默认：1 秒）

backoff_coefficient：后续延迟的倍增系数（默认：2.0）

maximum_interval：重试延迟的上限（默认：初始间隔的 100 倍）

maximum_attempts：最大重试次数（默认：无限）

non_retryable_error_types：不可重试的错误类型列表（默认：空）

根据失败类型匹配重试策略

对于瞬时失败（短暂的网络问题）：使用默认设置，或设置较低的 initial_interval 和 backoff_coefficient。

对于间歇性失败（速率限制）：增大 initial_interval 和 backoff_coefficient，拉长重试间隔。

对于对成本敏感的 API：设置 maximum_attempts 限制重试次数（少见——通常更倾向于使用超时控制）。

为不同的 Activity 使用不同的策略

可以为不同的 Activity 使用不同的重试策略，甚至可以为同一个 Activity 使用多种策略：

fast_retry = RetryPolicy(
    initial_interval=timedelta(seconds=1),
    backoff_coefficient=1.5,
)

slow_retry = RetryPolicy(
    initial_interval=timedelta(seconds=30),
    backoff_coefficient=3.0,
)

# Same Activity, different policies
await workflow.execute_activity(
    process_order,
    order,
    start_to_close_timeout=timedelta(seconds=10),
    retry_policy=fast_retry,
)

# Later, with different circumstances...
await workflow.execute_activity(
    process_order,
    order,
    start_to_close_timeout=timedelta(seconds=10),
    retry_policy=slow_retry,
)

不要使用工作流重试策略

与 Activity 不同，工作流默认不重试，而且通常也不应添加重试策略。工作流具有确定性，并非为易失败操作而设计。工作流失败通常意味着代码 Bug 或输入数据有误——重试整个工作流只会重复相同的逻辑，无法修复根本问题。

如果需要对特定的工作流操作实施重试逻辑，那么在工作流代码中实现，而不是使用工作流重试策略。

将特定错误标记为不可重试

如何使用 Temporal Python SDK 将特定错误标记为不可重试

有些失败是永久性的，无法通过重试解决。可以将这些失败标记为不可重试，以便快速失败，而不是等待超时。

在抛出 ApplicationError 时设置 non_retryable 标志：

from temporalio import activity
from temporalio.exceptions import ApplicationError

@activity.defn
async def process_payment(card_number: str, amount: float):
    if not is_valid_card_format(card_number):
        # Invalid format will never become valid through retries
        raise ApplicationError(
            f"Invalid credit card format: {card_number}",
            type="InvalidCardFormat",
            non_retryable=True,
        )

    if amount <= 0:
        # Invalid amount won't be fixed by retrying
        raise ApplicationError(
            f"Amount must be positive: {amount}",
            type="InvalidAmount",
            non_retryable=True,
        )

    # Process payment...

无论如何配置重试策略，设置 non_retryable=True 的 ApplicationError 都不重试。

不可重试错误可用于以下场景：

无效输入数据

业务规则违规

授权失败

需谨慎使用此功能。在大多数情况下，让重试策略根据时间或重试次数决定何时停止重试更优。

指定不可重试的错误类型

如何使用 Temporal Python SDK 在重试策略中指定不可重试的错误类型

有时，希望由工作流（调用方）决定哪些错误类型不应重试，而不是由 Activity（实现方）决定。

在重试策略中列出不应重试的错误类型：

from temporalio import workflow
from temporalio.common import RetryPolicy
from datetime import timedelta

@workflow.defn
class CheckoutWorkflow:
    @workflow.run
    async def run(self, payment_details):
        retry_policy = RetryPolicy(
            non_retryable_error_types=[
                "InvalidCardFormat",
                "InsufficientFunds",
                "AccountClosed",
            ]
        )

        try:
            result = await workflow.execute_activity(
                process_payment,
                payment_details,
                start_to_close_timeout=timedelta(seconds=30),
                retry_policy=retry_policy,
            )
            return result
        except ActivityError as e:
            workflow.logger.error(f"Payment failed: {e.cause}")
            # Handle the non-retryable error...

当 Activity 抛出 ApplicationError 时，Temporal 检查其 type 是否在 non_retryable_error_types 列表中。如果匹配，则 Activity 立即失败，而不重试。

何时使用每种方式

在 Activity 中设置 non_retryable=True：当 Activity 实现方知道某个错误永久性不可恢复时使用。对所有调用方强制执行该约束。

在重试策略中设置 non_retryable_error_types：当调用方希望根据自己的业务逻辑决定哪些错误不可恢复时使用。这样，不同工作流可以对同一 Activity 做出不同的决策。

使用 Saga 模式实现回滚逻辑

如何使用 Temporal Python SDK 实现 Saga 模式

Saga 模式用于协调一系列操作，其中每个操作都有对应的补偿操作，用于撤销其副作用。如果任何操作失败，则按相反顺序执行补偿操作，回滚之前的操作。

适用于以下多步骤流程：

电商结账（支付、库存、物流）

跨服务的分布式事务

多阶段数据更新

from temporalio import workflow
from temporalio.exceptions import ActivityError
from datetime import timedelta

@workflow.defn
class OrderWorkflow:
    @workflow.run
    async def run(self, order):
        compensations = []

        try:
            # Reserve inventory
            compensations.append({
                "activity": revert_inventory,
                "input": order
            })
            await workflow.execute_activity(
                reserve_inventory,
                order,
                start_to_close_timeout=timedelta(seconds=10),
            )

            # Charge payment
            compensations.append({
                "activity": refund_payment,
                "input": order
            })
            payment_id = await workflow.execute_activity(
                charge_payment,
                order,
                start_to_close_timeout=timedelta(seconds=10),
            )

            # Create shipment
            compensations.append({
                "activity": cancel_shipment,
                "input": payment_id
            })
            shipment_id = await workflow.execute_activity(
                create_shipment,
                order,
                start_to_close_timeout=timedelta(seconds=10),
            )

            return {"payment_id": payment_id, "shipment_id": shipment_id}

        except ActivityError as e:
            workflow.logger.error(f"Order failed: {e.cause}, rolling back...")

            # Execute compensations in reverse order
            for compensation in reversed(compensations):
                try:
                    await workflow.execute_activity(
                        compensation["activity"],
                        compensation["input"],
                        start_to_close_timeout=timedelta(seconds=10),
                    )
                except ActivityError as comp_err:
                    # Log compensation failure but continue with others
                    workflow.logger.error(f"Compensation failed: {comp_err.cause}")

            # Re-raise the original error
            raise ApplicationError(
                f"Order failed: {e.cause}",
                type="OrderFailed"
            )

关键要点：

在执行每个 Activity 前，将补偿操作添加到列表中

使用 reversed(compensations) 按正确顺序撤销操作

优雅地处理补偿失败（补偿操作本身也可能失败）

Temporal 负责管理所有状态和重试逻辑，使 Saga 实现更直接

理解 Temporal 失败类型

Temporal 使用专门的异常类型，表示不同的失败场景。所有异常都继承自 TemporalError。

请勿扩展 TemporalError 或其子类。使用 Temporal 提供的异常类型，可以确保：

跨进程和跨语言边界的行为一致性

与 Temporal 服务端的兼容性

通过 Protocol Buffer 正确地序列化

常见失败类型

ApplicationError：由代码抛出，用于表示应用程序特定的失败。这是唯一应手动抛出的 Temporal 异常。抛出 ApplicationError 时，可以选择提供 type 字符串，及将其标记为 non_retryable。

ActivityError：包装从 Activity 抛出的异常。cause 字段包含原始错误（ApplicationError、TimeoutError、CancelledError 等）。在工作流中捕获此异常，处理 Activity 失败。

TimeoutError：当 Activity 或工作流超过其配置的超时时间时发生。

CancelledError：由工作流、Activity 或定时器被取消而产生。可以捕获并忽略此异常，在取消后继续执行。

TerminatedError：当强制终止工作流执行时发生。

ChildWorkflowError：当子工作流执行失败时抛出。

WorkflowAlreadyStartedError：当尝试使用已在运行的 ID 启动工作流时抛出。

ServerError：用于 Temporal 服务端自身产生的异常（比如数据库故障）。

工作流任务失败与工作流执行失败的区别

工作流任务失败：当工作流代码抛出非 Temporal 类型的异常（比如 ValueError、TypeError 或非确定性错误）时发生。此类失败自动重试，能够在保留工作流状态的前提下，修复问题，重新部署。

工作流执行失败：当工作流代码抛出 Temporal 异常（比如 ApplicationError）时发生。此类失败使工作流进入“Failed”状态，且不触发自动重试。

以下是应导致工作流失败的永久性失败示例：

if distance.kilometers > MAX_DELIVERY_DISTANCE:
    # Retrying won't change the distance - this is permanent
    raise ApplicationError(
        "Customer lives outside service area",
        type="OutsideServiceArea"
    )

保护敏感信息

默认的失败转换器（Failure Converter）以纯文本形式复制异常信息和堆栈跟踪，并且在 Web UI 中可见。如果异常包含敏感信息，那么配置自定义失败转换器加密该数据。