Lambda 耐用函數的最佳實務

耐用函數使用重播型執行模型，需要與傳統 Lambda 函數不同的模式。遵循這些最佳實務來建置可靠且符合成本效益的工作流程。

寫入確定性程式碼

在重播期間，您的函數會從頭開始執行，並且必須遵循與原始執行相同的執行路徑。持久性操作之外的程式碼必須具有決定性，在相同的輸入下產生相同的結果。

以步驟包裝非確定性操作：

• 隨機數產生和 UUIDs

• 目前時間或時間戳記

• 外部 API 呼叫和資料庫查詢

• 檔案系統操作

TypeScript

import { withDurableExecution, DurableContext } from '@aws/durable-execution-sdk-js';
import { randomUUID } from 'crypto';

export const handler = withDurableExecution(
  async (event: any, context: DurableContext) => {
    // Generate transaction ID inside a step
    const transactionId = await context.step('generate-transaction-id', async () => {
      return randomUUID();
    });
    
    // Use the same ID throughout execution, even during replay
    const payment = await context.step('process-payment', async () => {
      return processPayment(event.amount, transactionId);
    });
    
    return { statusCode: 200, transactionId, payment };
  }
);
          

Python

from aws_durable_execution_sdk_python import durable_execution, DurableContext
import uuid

@durable_execution
def handler(event, context: DurableContext):
    # Generate transaction ID inside a step
    transaction_id = context.step(
        lambda _: str(uuid.uuid4()),
        name='generate-transaction-id'
    )
    
    # Use the same ID throughout execution, even during replay
    payment = context.step(
        lambda _: process_payment(event['amount'], transaction_id),
        name='process-payment'
    )
    
    return {'statusCode': 200, 'transactionId': transaction_id, 'payment': payment}
          

Important (重要)

請勿使用全域變數或關閉在步驟之間共用狀態。透過傳回值傳遞資料。重播期間全域狀態中斷，因為步驟會傳回快取的結果，但全域變數會重設。

避免關閉變動：在關閉中擷取的變數可能會在重播期間遺失變動。步驟會傳回快取的結果，但步驟外的變數更新不會重播。

TypeScript

// ❌ WRONG: Mutations lost on replay
export const handler = withDurableExecution(async (event, context) => {
  let total = 0;
  
  for (const item of items) {
    await context.step(async () => {
      total += item.price; // ⚠️ Mutation lost on replay!
      return saveItem(item);
    });
  }
  
  return { total }; // Inconsistent value!
});

// ✅ CORRECT: Accumulate with return values
export const handler = withDurableExecution(async (event, context) => {
  let total = 0;
  
  for (const item of items) {
    total = await context.step(async () => {
      const newTotal = total + item.price;
      await saveItem(item);
      return newTotal; // Return updated value
    });
  }
  
  return { total }; // Consistent!
});

// ✅ EVEN BETTER: Use map for parallel processing
export const handler = withDurableExecution(async (event, context) => {
  const results = await context.map(
    items,
    async (ctx, item) => {
      await ctx.step(async () => saveItem(item));
      return item.price;
    }
  );
  
  const total = results.getResults().reduce((sum, price) => sum + price, 0);
  return { total };
});
          

Python

# ❌ WRONG: Mutations lost on replay
@durable_execution
def handler(event, context: DurableContext):
    total = 0
    
    for item in items:
        context.step(
            lambda _: save_item_and_mutate(item, total),  # ⚠️ Mutation lost on replay!
            name=f'save-item-{item["id"]}'
        )
    
    return {'total': total}  # Inconsistent value!

# ✅ CORRECT: Accumulate with return values
@durable_execution
def handler(event, context: DurableContext):
    total = 0
    
    for item in items:
        total = context.step(
            lambda _: save_item_and_return_total(item, total),
            name=f'save-item-{item["id"]}'
        )
    
    return {'total': total}  # Consistent!

# ✅ EVEN BETTER: Use map for parallel processing
@durable_execution
def handler(event, context: DurableContext):
    def process_item(ctx, item):
        ctx.step(lambda _: save_item(item))
        return item['price']
    
    results = context.map(items, process_item)
    total = sum(results.get_results())
    
    return {'total': total}
          

等冪性的設計

由於重試或重播，操作可能會執行多次。非等冪性操作會導致重複的副作用，例如向客戶收費兩次或傳送多封電子郵件。

使用等冪符記：在步驟中產生符記，並將其包含在外部 API 呼叫中，以防止重複操作。

TypeScript

import { withDurableExecution, DurableContext } from '@aws/durable-execution-sdk-js';

export const handler = withDurableExecution(
  async (event: any, context: DurableContext) => {
    // Generate idempotency token once
    const idempotencyToken = await context.step('generate-idempotency-token', async () => {
      return crypto.randomUUID();
    });
    
    // Use token to prevent duplicate charges
    const charge = await context.step('charge-payment', async () => {
      return paymentService.charge({
        amount: event.amount,
        cardToken: event.cardToken,
        idempotencyKey: idempotencyToken
      });
    });
    
    return { statusCode: 200, charge };
  }
);
          

Python

from aws_durable_execution_sdk_python import durable_execution, DurableContext
import uuid

@durable_execution
def handler(event, context: DurableContext):
    # Generate idempotency token once
    idempotency_token = context.step(
        lambda _: str(uuid.uuid4()),
        name='generate-idempotency-token'
    )
    
    # Use token to prevent duplicate charges
    def charge_payment(_):
        return payment_service.charge(
            amount=event['amount'],
            card_token=event['cardToken'],
            idempotency_key=idempotency_token
        )
    
    charge = context.step(charge_payment, name='charge-payment')
    
    return {'statusCode': 200, 'charge': charge}
          

使用at-most-once語意：對於絕不能重複的關鍵操作 （金融交易、庫存扣除），請設定at-most-once的執行模式。

TypeScript

// Critical operation that must not duplicate
await context.step('deduct-inventory', async () => {
  return inventoryService.deduct(event.productId, event.quantity);
}, {
  executionMode: 'AT_MOST_ONCE_PER_RETRY'
});
          

Python

# Critical operation that must not duplicate
context.step(
    lambda _: inventory_service.deduct(event['productId'], event['quantity']),
    name='deduct-inventory',
    config=StepConfig(execution_mode='AT_MOST_ONCE_PER_RETRY')
)
          

資料庫冪等性：使用check-before-write模式、條件更新或 upsert 操作，以防止重複記錄。

有效率地管理狀態

每個檢查點都會將狀態儲存為持久性儲存。大型狀態物件會增加成本、緩慢檢查點和影響效能。僅儲存必要的工作流程協調資料。

保持最小狀態：

• 存放 IDs和參考，而非完整物件

• 視需要在步驟內擷取詳細資訊

• 針對大型資料使用 Amazon S3 或 DynamoDB，以 狀態傳遞參考

• 避免在步驟之間傳遞大型承載

TypeScript

import { withDurableExecution, DurableContext } from '@aws/durable-execution-sdk-js';

export const handler = withDurableExecution(
  async (event: any, context: DurableContext) => {
    // Store only the order ID, not the full order object
    const orderId = event.orderId;
    
    // Fetch data within each step as needed
    await context.step('validate-order', async () => {
      const order = await orderService.getOrder(orderId);
      return validateOrder(order);
    });
    
    await context.step('process-payment', async () => {
      const order = await orderService.getOrder(orderId);
      return processPayment(order);
    });
    
    return { statusCode: 200, orderId };
  }
);
          

Python

from aws_durable_execution_sdk_python import durable_execution, DurableContext

@durable_execution
def handler(event, context: DurableContext):
    # Store only the order ID, not the full order object
    order_id = event['orderId']
    
    # Fetch data within each step as needed
    context.step(
        lambda _: validate_order(order_service.get_order(order_id)),
        name='validate-order'
    )
    
    context.step(
        lambda _: process_payment(order_service.get_order(order_id)),
        name='process-payment'
    )
    
    return {'statusCode': 200, 'orderId': order_id}
          

設計有效的步驟

步驟是耐用函數中的基本工作單位。精心設計的步驟可讓工作流程更易於理解、偵錯和維護。

步驟設計原則：

• 使用描述性名稱 - 類似 的名稱，validate-order而不是step1讓日誌和錯誤更容易理解

• 將名稱保持靜態 - 請勿將動態名稱與時間戳記或隨機值搭配使用。步驟名稱必須具有決定性才能重播

• 平衡精細程度 - 將複雜的操作分解為重點步驟，但避免增加檢查點負荷的過小步驟

• 群組相關操作 - 應該一起成功或失敗的操作屬於同一步驟

有效率地使用等待操作

等待操作會暫停執行，而不會耗用資源或產生成本。使用它們，而不是讓 Lambda 保持執行狀態。

以時間為基礎的等待：使用 context.wait() 處理延遲，而非 setTimeout或 sleep。

外部回呼：等待外部系統context.waitForCallback()時使用 。一律設定逾時以防止無限期等待。

輪詢：使用 context.waitForCondition()搭配指數退避來輪詢外部服務，而不會讓它們過多。

TypeScript

// Wait 24 hours without cost
await context.wait({ seconds: 86400 });

// Wait for external callback with timeout
const result = await context.waitForCallback(
  'external-job',
  async (callbackId) => {
    await externalService.submitJob({
      data: event.data,
      webhookUrl: `https://api.example.com/callbacks/${callbackId}`
    });
  },
  { timeout: { seconds: 3600 } }
);
          

Python

# Wait 24 hours without cost
context.wait(86400)

# Wait for external callback with timeout
result = context.wait_for_callback(
    lambda callback_id: external_service.submit_job(
        data=event['data'],
        webhook_url=f'https://api.example.com/callbacks/{callback_id}'
    ),
    name='external-job',
    config=WaitForCallbackConfig(timeout_seconds=3600)
)
          

其他考量

錯誤處理：重試暫時性故障，例如網路逾時和速率限制。請勿重試永久失敗，例如無效的輸入或身分驗證錯誤。使用適當的最大嘗試次數和退避率來設定重試策略。如需詳細範例，請參閱錯誤處理和重試。

效能：透過儲存參考而非完整承載，將檢查點大小降至最低。使用 context.parallel()和 context.map() 同時執行獨立操作。批次相關操作，以減少檢查點額外負荷。

版本控制：使用版本編號或別名叫用函數，將執行釘選到特定的程式碼版本。確保新的程式碼版本可以處理較舊版本的狀態。請勿重新命名步驟或以中斷重播的方式變更其行為。

序列化：將 JSON 相容類型用於操作輸入和結果。將日期轉換為 ISO 字串，並將自訂物件轉換為純物件，然後再將其傳遞至持久的操作。

監控：啟用具有執行 IDs結構化記錄。針對錯誤率和執行持續時間設定 CloudWatch 警示。使用追蹤來識別瓶頸。如需詳細指引，請參閱監控和偵錯。

測試：測試快樂路徑、錯誤處理和重播行為。回呼和等待的測試逾時案例。使用本機測試來縮短反覆運算時間。如需詳細指引，請參閱測試耐用函數。

要避免的常見錯誤：不要巢狀context.step()化呼叫，請改用子內容。在步驟中包裝非確定性操作。一律設定回呼的逾時。平衡步驟精細程度與檢查點額外負荷。儲存參考，而不是處於 狀態的大型物件。

其他資源

• Python SDK 文件 - 完整的 API 參考、測試模式和進階範例

• TypeScript SDK 文件 - 完整的 API 參考、測試模式和進階範例