Evitar que los reintentos del flujo repitan un efecto de negocio

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.

Step Functions coordina tareas de Lambda; el procesador proporcionado guarda pedidos en DynamoDB y lee sus resúmenes. El acceso mediante CLI ya está configurado; no necesitas una cuenta personal de AWS ni claves de acceso.

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

Vista general del laboratorio

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.

Usa AWS View en la parte superior 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.

La protección debe estar en la escritura de negocio porque un nombre de ejecución no protege un pedido compartido por varias ejecuciones. Considera completado solo un elemento existente idéntico. Reutilizar el mismo ID con contenido distinto es un conflicto que debe notificarse, no un permiso para sobrescribirlo.

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

Aplica el nuevo paquete de código de la función:

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

Guarda el valor devuelto en WORKER_ARN:

WORKER_ARN=$(aws lambda get-function-configuration \
  --function-name labex-ev05-worker \
  --query FunctionArn \
  --output text)
echo "$WORKER_ARN"

Consulta el rol y la configuración de entorno de la función:

aws lambda get-function-configuration \
  --function-name labex-ev05-worker \
  --query '{Name:FunctionName,Role:Role,Runtime:Runtime,Timeout:Timeout}'

Consulta las máquinas de estados restantes:

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

Guarda el valor devuelto en ROLE_ARN:

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)
echo "$ROLE_ARN"

cat > workflow-invoke.json <<EOF
{
  "Version": "2012-10-17",
  "Statement": [
    {
      "Effect": "Allow",
      "Action": "lambda:InvokeFunction",
      "Resource": "$WORKER_ARN"
    }
  ]
}
EOF

Aplica la política de permisos del rol de la aplicación:

aws iam put-role-policy \
  --role-name labex-ev05-workflow-role \
  --policy-name InvokeWorker \
  --policy-document file://workflow-invoke.json

Consulta la política de permisos del rol de la aplicación:

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

Escribe la configuración en workflow.json:

jq --arg worker "$WORKER_NAME" '.States.StoreOrder.Parameters.FunctionName=$worker | .States.ReadSummary.Parameters.FunctionName=$worker' workflow-template.json > workflow.json

Guarda el valor devuelto en MACHINE_ARN:

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)
echo "$MACHINE_ARN"

Consulta la definición y el rol de la máquina de estados:

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)
echo "$FIRST_ARN"

Consulta esta ejecución hasta que deje el estado RUNNING:

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

Consulta el estado, la salida y los errores de la ejecución:

aws stepfunctions describe-execution \
  --execution-arn "$FIRST_ARN" \
  --query '{Status:status,Output:output}'

Consulta el historial real de ejecución del flujo:

aws stepfunctions get-execution-history \
  --execution-arn "$FIRST_ARN" \
  --query 'events[?type==`TaskFailed` || type==`TaskSucceeded`].{Type:type,Error:taskFailedEventDetails.error,Output:taskSucceededEventDetails.output}'

Consulta el elemento realmente almacenado en la base de datos:

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)
echo "$REPEAT_ARN"

Consulta esta ejecución hasta que deje el estado RUNNING:

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

Consulta el estado, la salida y los errores de la ejecución:

aws stepfunctions describe-execution \
  --execution-arn "$REPEAT_ARN" \
  --query '{Status:status,Output:output}'

Consulta el historial real de ejecución del flujo:

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)
echo "$NEW_ARN"

Consulta esta ejecución hasta que deje el estado RUNNING:

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

Consulta el estado, la salida y los errores de la ejecución:

aws stepfunctions describe-execution \
  --execution-arn "$NEW_ARN" \
  --query '{Status:status,Output:output}'

Consulta los pedidos realmente almacenados:

aws dynamodb scan --table-name labex-ev05-orders --query Items

Consulta los mensajes reales de los registros de ejecución:

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.

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"

Elimina solo la política de permisos del rol de la aplicación:

aws iam delete-role-policy --role-name labex-ev05-workflow-role --policy-name InvokeWorker

Elimina solo el rol de ejecución del flujo del ejercicio:

aws iam delete-role --role-name labex-ev05-workflow-role

Elimina solo el elemento realmente almacenado en la base de datos:

aws dynamodb delete-item \
  --table-name labex-ev05-orders \
  --key '{"id":{"S":"protected-order"}}'

Elimina solo el elemento realmente almacenado en la base de datos:

aws dynamodb delete-item \
  --table-name labex-ev05-attempts \
  --key '{"id":{"S":"protected-order"}}'

Elimina solo el elemento realmente almacenado en la base de datos:

aws dynamodb delete-item \
  --table-name labex-ev05-orders \
  --key '{"id":{"S":"different-order"}}'

Elimina solo el elemento realmente almacenado en la base de datos:

aws dynamodb delete-item \
  --table-name labex-ev05-attempts \
  --key '{"id":{"S":"different-order"}}'

Elimina solo el grupo de registros de la aplicación indicado:

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

Consulta los roles IAM restantes:

aws iam list-roles --query 'Roles[].RoleName'

Consulta los pedidos realmente almacenados:

aws dynamodb scan --table-name labex-ev05-orders --query Items

Consulta los grupos de registros restantes:

aws logs describe-log-groups --query logGroups

Consulta el registro de referencia ajeno al ejercicio:

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.

Para ampliar este tema, consulta la referencia de AWS: Escrituras condicionales DynamoDB y protección contra sobrescritura; Retry, retroceso y manejo de errores Catch.