LabEx
Únete

Evitar que los reintentos del flujo repitan un efecto de negocio

AWSBeginner
Practicar Ahora

Introducción

Un procesador guarda un pedido, pero falla antes de que su resultado llegue al flujo. Protegerás esa escritura para que un reintento o una ejecución repetida conserve el pedido original, mientras un pedido distinto todavía puede completarse correctamente.

Completa primero Reintentar un paso fallido y gestionar errores permanentes, Evitar pedidos duplicados con escrituras condicionales y Configurar y diagnosticar una función Lambda. Esta VM independiente proporciona un procesador sin protección; tú añades la protección.

Relación con las certificaciones

Este laboratorio ofrece práctica para los siguientes temas de examen.

Proteger la escritura de negocio con una clave estable

En este paso, despliega un procesador que trate un pedido idéntico repetido como una escritura ya completada.

dos ejecuciones y un pedido de negocio

Distintos nombres de ejecución pueden referirse al mismo pedido de negocio. Un conflicto de escritura condicional idéntica devuelve un resultado de duplicado sin sustituir ese pedido.

Usa AWS View junto a Terminal para comparar las consultas CLI con los recursos y resultados reales de este laboratorio. Conserva los datos de referencia proporcionados.

El ID de pedido es una clave de negocio: identifica el pedido previsto incluso cuando difieren los intentos de tarea o los nombres de ejecución del flujo. attribute_not_exists(id) permite únicamente el primer PutItem. DynamoDB evalúa esa condición de forma atómica. ReturnValuesOnConditionCheckFailure:ALL_OLD proporciona el elemento existente cuando un reintento pierde la carrera de la condición. Devuelve un resultado de duplicado solo si ese elemento existente es igual al pedido previsto; el contenido en conflicto debe seguir fallando en lugar de sustituir silenciosamente un pedido.

El código inicial proporcionado tiene una escritura incondicional. Sustitúyelo por el controlador protegido completo siguiente. Usa el entorno SDK configurado, sin credenciales personales. Los intentos de diagnóstico registran las invocaciones reales por separado de las escrituras de negocio. En el modo sintético after-write, el primer intento lanza un error después de la escritura nativa, creando la incertidumbre que debe gestionar un reintento. El modo summary lee el elemento realmente guardado.

Un documento de entrada entre comillas escribe el archivo Python literal. El controlador Lambda existente sigue siendo app.handler.

cd /home/labex/project
cat > app.py <<'PYTHON'
import json
import os
import boto3
from botocore.exceptions import ClientError

class TransientOrderError(Exception):
    pass
class InvalidOrder(Exception):
    pass

def handler(event, context):
    print('WORKFLOW '+json.dumps(event,sort_keys=True))
    db=boto3.client('dynamodb')
    if event.get('stage')=='summary':
        item=db.get_item(
            TableName=os.environ['ORDERS_TABLE'],
            Key={'id':{'S':event['id']}},
        )['Item']
        result={'id':event['id'],'total_cents':int(item['total_cents']['N']),'completed':True}
        print('RESULT '+json.dumps(result,sort_keys=True))
        return result
    if event.get('mode')=='permanent':
        raise InvalidOrder('Order cannot be completed')
    quantity=event['quantity']
    if isinstance(quantity,bool) or not isinstance(quantity,int) or not 1<=quantity<=10:
        raise InvalidOrder('Invalid quantity')
    attempt=db.update_item(
        TableName=os.environ['ATTEMPTS_TABLE'],
        Key={'id':{'S':event['id']}},
        UpdateExpression='ADD attempts :one',
        ExpressionAttributeValues={':one':{'N':'1'}},
        ReturnValues='UPDATED_NEW',
    )['Attributes']['attempts']['N']
    if event.get('mode')=='flaky' and attempt=='1':
        raise TransientOrderError('Dependency temporarily unavailable')
    item={'id':{'S':event['id']},'quantity':{'N':str(quantity)},'total_cents':{'N':str(quantity*250+100)}}
    duplicate=False
    try:
        db.put_item(
            TableName=os.environ['ORDERS_TABLE'],
            Item=item,
            ConditionExpression='attribute_not_exists(id)',
            ReturnValuesOnConditionCheckFailure='ALL_OLD',
        )
    except ClientError as error:
        if (
            error.response['Error']['Code']!='ConditionalCheckFailedException'
            or error.response.get('Item')!=item
        ):
            raise
        duplicate=True
    if event.get('mode')=='after-write' and attempt=='1':
        raise TransientOrderError('Result delivery failed after the write')
    result={'id':event['id'],'quantity':quantity,'total_cents':quantity*250+100,'duplicate':duplicate}
    print('RESULT '+json.dumps(result,sort_keys=True))
    return result
PYTHON

El try realiza la escritura condicional nativa. Solo una ConditionalCheckFailedException real con un elemento anterior idéntico se trata como duplicado; los demás errores SDK se vuelven a lanzar. La excepción temporal ocurre después de esa decisión de escritura, por lo que el flujo reintentará realmente un trabajo cuyo elemento de negocio ya existe.

Empaqueta el archivo usando zip -j, que omite las rutas de directorios, y actualiza el procesador proporcionado con --zip-file fileb:// para los bytes binarios del archivo comprimido. El CodeSha256 de la respuesta identifica el archivo desplegado. El despliegue por sí solo no demuestra protección contra duplicados; el paso de ejecución la comprobará.

zip -j function.zip app.py
aws lambda update-function-code \
  --function-name labex-ev05-worker \
  --zip-file fileb://function.zip \
  --query CodeSha256 \
  --output text

Ejecuta la comprobación de despliegue antes de crear ejecuciones.

Conectar un flujo con permisos limitados y reintentos limitados

En este paso, crea un rol y una máquina independientes de flujo que puedan reintentar el fallo del procesador posterior a la escritura.

Step Functions necesita confianza para el servicio states y un permiso InvokeFunction separado sobre la función exacta. El procesador conserva sus propios permisos DynamoDB; el rol del flujo solo lo invoca. Las asignaciones de shell guardan identificadores devueltos. --query y --output text seleccionan valores reutilizables, file:// lee el JSON literal de confianza y el shell inserta el ARN del procesador en el documento de permisos.

cd /home/labex/project
WORKER_NAME=labex-ev05-worker
WORKER_ARN=$(aws lambda get-function-configuration \
  --function-name labex-ev05-worker \
  --query FunctionArn \
  --output text)
aws lambda get-function-configuration \
  --function-name labex-ev05-worker \
  --query '{Name:FunctionName,Role:Role,Runtime:Runtime,Timeout:Timeout}'
aws stepfunctions list-state-machines

cat > workflow-trust.json <<'JSON'
{
  "Version": "2012-10-17",
  "Statement": [
    {
      "Effect": "Allow",
      "Principal": {
        "Service": "states.amazonaws.com"
      },
      "Action": "sts:AssumeRole"
    }
  ]
}
JSON
ROLE_ARN=$(aws iam create-role \
  --role-name labex-ev05-workflow-role \
  --assume-role-policy-document file://workflow-trust.json \
  --query Role.Arn \
  --output text)

cat > workflow-invoke.json <<EOF
{
  "Version": "2012-10-17",
  "Statement": [
    {
      "Effect": "Allow",
      "Action": "lambda:InvokeFunction",
      "Resource": "$WORKER_ARN"
    }
  ]
}
EOF
aws iam put-role-policy \
  --role-name labex-ev05-workflow-role \
  --policy-name InvokeWorker \
  --policy-document file://workflow-invoke.json
aws iam get-role-policy --role-name labex-ev05-workflow-role --policy-name InvokeWorker

La lista vacía de máquinas confirma que no se reutiliza ningún flujo de una VM anterior. El permiso usa el ARN de función, mientras cada Task usa su nombre ordinario de función. Retry solo trata TransientOrderError con un intervalo de un segundo, espera creciente que se duplica y un máximo de dos reintentos después del primer intento; Catch dirige InvalidOrder a un Fail explícito. ResultSelector conserva el Payload real, ResultPath lo conserva en saved y OutputPath devuelve el resumen real.

Un documento de entrada entre comillas escribe ASL literal. jq --arg sustituye los marcadores del procesador antes de crear la máquina.

cat > workflow-template.json <<'JSON'
{
  "StartAt": "CheckQuantity",
  "States": {
    "CheckQuantity": {
      "Type": "Choice",
      "Choices": [
        {
          "Variable": "$.quantity",
          "NumericGreaterThan": 0,
          "Next": "StoreOrder"
        }
      ],
      "Default": "Rejected"
    },
    "StoreOrder": {
      "Type": "Task",
      "Resource": "arn:aws:states:::lambda:invoke",
      "Parameters": {
        "FunctionName": "WORKER_NAME",
        "Payload.$": "$"
      },
      "ResultSelector": {
        "result.$": "$.Payload"
      },
      "ResultPath": "$.saved",
      "Retry": [
        {
          "ErrorEquals": [
            "TransientOrderError"
          ],
          "IntervalSeconds": 1,
          "BackoffRate": 2,
          "MaxAttempts": 2
        }
      ],
      "Catch": [
        {
          "ErrorEquals": [
            "InvalidOrder"
          ],
          "Next": "Rejected",
          "ResultPath": "$.failure"
        }
      ],
      "Next": "ReadSummary"
    },
    "ReadSummary": {
      "Type": "Task",
      "Resource": "arn:aws:states:::lambda:invoke",
      "Parameters": {
        "FunctionName": "WORKER_NAME",
        "Payload": {
          "stage": "summary",
          "id.$": "$.saved.result.id"
        }
      },
      "OutputPath": "$.Payload",
      "End": true
    },
    "Rejected": {
      "Type": "Fail",
      "Error": "OrderRejected",
      "Cause": "Order could not be completed"
    }
  }
}
JSON
jq --arg worker "$WORKER_NAME" '.States.StoreOrder.Parameters.FunctionName=$worker | .States.ReadSummary.Parameters.FunctionName=$worker' workflow-template.json > workflow.json
MACHINE_ARN=$(aws stepfunctions create-state-machine \
  --name labex-ev05-orders \
  --type STANDARD \
  --role-arn "$ROLE_ARN" \
  --definition file://workflow.json \
  --query stateMachineArn \
  --output text)
aws stepfunctions describe-state-machine \
  --state-machine-arn "$MACHINE_ARN" \
  --query '{Name:name,Definition:definition}'

La definición nativa enumera Retry/Catch selectivos y dos tareas reales. AWS View muestra la máquina sin ejecuciones ni pedidos todavía. Ejecuta la comprobación de configuración del flujo.

Demostrar una escritura de negocio entre reintentos y nuevas ejecuciones

En este paso, ejecuta un fallo posterior a la escritura, repite el mismo pedido y crea un pedido distinto.

Los nombres de ejecución identifican ejecuciones del flujo; los IDs de pedido identifican efectos de negocio. La primera ejecución usa el modo after-write. Un bucle con límite lee el estado hasta que deja de estar RUNNING; $(...) captura la salida y break sale del bucle. Inspecciona la ejecución si sigue RUNNING después del bucle.

FIRST_ARN=$(aws stepfunctions start-execution \
  --state-machine-arn "$MACHINE_ARN" \
  --name after-write-order \
  --input '{"id":"protected-order","quantity":4,"mode":"after-write"}' \
  --query executionArn \
  --output text)
for attempt in $(seq 1 60); do
  STATUS=$(aws stepfunctions describe-execution \
    --execution-arn "$FIRST_ARN" \
    --query status \
    --output text)
  if test "$STATUS" != RUNNING; then break; fi
  sleep 2
done
aws stepfunctions describe-execution \
  --execution-arn "$FIRST_ARN" \
  --query '{Status:status,Output:output}'
aws stepfunctions get-execution-history \
  --execution-arn "$FIRST_ARN" \
  --query 'events[?type==`TaskFailed` || type==`TaskSucceeded`].{Type:type,Error:taskFailedEventDetails.error,Output:taskSucceededEventDetails.output}'
aws dynamodb get-item \
  --table-name labex-ev05-orders \
  --key '{"id":{"S":"protected-order"}}' \
  --query Item

El primer TaskFailed es TransientOrderError después de la escritura real. El reintento correcto devuelve duplicate:true; el resumen se completa con total 1100. El único elemento protected-order tiene cantidad 4 y total 1100. El reintento de tarea se completó correctamente sin un segundo PutItem aceptado para esa clave de negocio.

Inicia un nombre de ejecución nuevo con el mismo ID y valores de pedido:

REPEAT_ARN=$(aws stepfunctions start-execution \
  --state-machine-arn "$MACHINE_ARN" \
  --name repeated-order \
  --input '{"id":"protected-order","quantity":4,"mode":"after-write"}' \
  --query executionArn \
  --output text)
for attempt in $(seq 1 60); do
  STATUS=$(aws stepfunctions describe-execution \
    --execution-arn "$REPEAT_ARN" \
    --query status \
    --output text)
  if test "$STATUS" != RUNNING; then break; fi
  sleep 2
done
aws stepfunctions describe-execution \
  --execution-arn "$REPEAT_ARN" \
  --query '{Status:status,Output:output}'
aws stepfunctions get-execution-history \
  --execution-arn "$REPEAT_ARN" \
  --query 'events[?type==`TaskSucceeded`].taskSucceededEventDetails.output'

Esta nueva ejecución también devuelve un resultado de almacenamiento duplicado y lee el resumen original de 1100. Un nombre de ejecución nuevo no convierte el pedido en una nueva solicitud de negocio. Los fallos reales de condición conservan el elemento original en lugar de sobrescribirlo.

Usa un ID de pedido distinto para demostrar que la protección no rechaza trabajo independiente:

NEW_ARN=$(aws stepfunctions start-execution \
  --state-machine-arn "$MACHINE_ARN" \
  --name different-order \
  --input '{"id":"different-order","quantity":1,"mode":"normal"}' \
  --query executionArn \
  --output text)
for attempt in $(seq 1 60); do
  STATUS=$(aws stepfunctions describe-execution \
    --execution-arn "$NEW_ARN" \
    --query status \
    --output text)
  if test "$STATUS" != RUNNING; then break; fi
  sleep 2
done
aws stepfunctions describe-execution \
  --execution-arn "$NEW_ARN" \
  --query '{Status:status,Output:output}'
aws dynamodb scan --table-name labex-ev05-orders --query Items
aws logs filter-log-events \
  --log-group-name /aws/lambda/labex-ev05-worker \
  --query 'events[].message'

El pedido distinto se completa correctamente con 1/350. Hay dos elementos de negocio, aunque se realizaron siete llamadas reales al procesador entre intentos de almacenamiento y lecturas de resumen. Los registros y los resultados nativos de tareas muestran decisiones de duplicado para la clave original. AWS View muestra tres resúmenes correctos y ambos pedidos conservados.

El siguiente ejemplo muestra resúmenes reales del reintento y de la nueva ejecución, junto con el pedido original conservado y el pedido nuevo separado.

AWS View muestra el pedido original conservado y un pedido de negocio distinto Esta es una escritura idempotente para la solicitud idéntica comprobada, no una promesa de que las tareas se ejecuten exactamente una vez.

Ejecuta la comprobación de protección de negocio.

Eliminar los recursos del flujo y los resultados sintéticos

En este paso, elimina tu máquina completada, el rol del flujo, el pedido y los registros, conservando los recursos preparados proporcionados.

Las tres ejecuciones han terminado. Eliminar la máquina la retira de la lista de máquinas activas. Elimina la política del rol propio antes de eliminar el rol y después elimina el pedido sintético y el grupo de registros del procesador creados por tu ejecución.

aws stepfunctions delete-state-machine --state-machine-arn "$MACHINE_ARN"
aws iam delete-role-policy --role-name labex-ev05-workflow-role --policy-name InvokeWorker
aws iam delete-role --role-name labex-ev05-workflow-role
aws dynamodb delete-item \
  --table-name labex-ev05-orders \
  --key '{"id":{"S":"protected-order"}}'
aws dynamodb delete-item \
  --table-name labex-ev05-attempts \
  --key '{"id":{"S":"protected-order"}}'
aws dynamodb delete-item \
  --table-name labex-ev05-orders \
  --key '{"id":{"S":"different-order"}}'
aws dynamodb delete-item \
  --table-name labex-ev05-attempts \
  --key '{"id":{"S":"different-order"}}'
aws logs delete-log-group --log-group-name /aws/lambda/labex-ev05-worker

Lee inventarios correctos para demostrar qué permanece:

aws stepfunctions list-state-machines
aws iam list-roles --query 'Roles[].RoleName'
aws dynamodb scan --table-name labex-ev05-orders --query Items
aws logs describe-log-groups --query logGroups
aws dynamodb scan --table-name labex-ev05-reference --query Items

No hay máquinas activas, pedidos ni grupos de registros. Solo permanece el rol del procesador proporcionado y el elemento de referencia está sin cambios. Conserva el procesador y las tablas proporcionados: la responsabilidad de sus recursos preparados difiere de la de los recursos y el flujo que creaste. Los errores de red o autenticación nunca demuestran la eliminación.

Elimina los archivos ordinarios creados en este laboratorio:

rm -f app.py function.zip workflow-trust.json workflow-invoke.json workflow-template.json workflow.json

AWS View muestra máquinas, ejecuciones y pedidos vacíos y la referencia conservada. Ejecuta la comprobación de limpieza antes de finalizar la VM.

Resumen

Protegiste las escrituras nativas de negocio con una clave de pedido estable, trataste únicamente conflictos condicionales idénticos como duplicados y comprobaste un fallo real después de la primera escritura. El reintento y una nueva ejecución del flujo conservaron el pedido original; una clave de negocio distinta creó su propio resultado. Verificaste los historiales y los datos nativos antes de eliminar los recursos propios del flujo y los resultados sintéticos.

Relacionado AWS Cursos
Fundamentos de AWS para principiantes

Fundamentos de AWS para principiantes

aws
AWS S3 para principiantes

AWS S3 para principiantes

aws
AWS IAM para principiantes

AWS IAM para principiantes

aws