LabEx
Junte-se

Impeça que Novas Tentativas do Fluxo Repitam um Efeito de Negócio

AWSBeginner
Pratique Agora

Introdução

Uma função de processamento salva um pedido, mas falha antes que seu resultado chegue ao fluxo. Você protegerá essa gravação para que uma nova tentativa ou execução repetida preserve o pedido original, enquanto um pedido diferente ainda pode ser bem-sucedido.

Conclua primeiro Repita uma Etapa com Falha e Trate Erros Permanentes, Impeça pedidos duplicados com gravações condicionais e Configure e Diagnostique uma Função Lambda. Esta VM independente fornece uma função de processamento sem proteção; você adiciona a proteção.

Relação com as certificações

Este laboratório oferece prática nos seguintes tópicos de exame.

Proteja a Gravação de Negócio com uma Chave Estável

Nesta etapa, implante uma função de processamento que trate um pedido idêntico repetido como uma gravação já concluída.

Duas execuções, um pedido de negócio

Nomes de execução diferentes podem se referir ao mesmo pedido de negócio. Um conflito de gravação condicional com conteúdo idêntico retorna um resultado de duplicata sem substituir esse pedido.

Use AWS View ao lado do Terminal para comparar as consultas da CLI com os recursos e resultados reais deste laboratório. Preserve os dados de referência fornecidos.

O ID do pedido é uma chave de negócio: identifica o pedido pretendido mesmo quando as tentativas de tarefas ou os nomes de execução do fluxo diferem. attribute_not_exists(id) permite apenas o primeiro PutItem. O DynamoDB avalia essa condição atomicamente. ReturnValuesOnConditionCheckFailure:ALL_OLD fornece o item existente quando uma nova tentativa perde a disputa pela condição. Retorne um resultado de duplicata apenas se esse item existente for igual ao pedido pretendido; conteúdo conflitante deve continuar causando falha, em vez de substituir um pedido silenciosamente.

A versão inicial fornecida tem uma gravação incondicional. Substitua-a pelo manipulador completo protegido abaixo. Ele usa o ambiente configurado do SDK, não credenciais pessoais. Tentativas de diagnóstico acompanham invocações reais separadamente das gravações de negócio. No modo sintético after-write, a primeira tentativa gera um erro após a gravação nativa, criando a incerteza que uma nova tentativa deve tratar. O modo summary lê o item real salvo.

Um here-document com marcador entre aspas grava o arquivo Python literal. O manipulador Lambda existente continua sendo 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

O try realiza a gravação condicional nativa. Apenas uma ConditionalCheckFailedException real com um item anterior idêntico é tratada como duplicata; outros erros do SDK são propagados novamente. A exceção temporária ocorre após essa decisão de gravação, portanto o fluxo realmente tentará novamente um trabalho cujo item de negócio já existe.

Empacote o arquivo usando zip -j, que omite caminhos de diretórios, e atualize a função de processamento fornecida com --zip-file fileb:// para os bytes do arquivo binário compactado. O CodeSha256 da resposta identifica o arquivo implantado. A implantação, por si só, não comprova proteção contra duplicatas; a etapa de execução a testará.

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

Execute a verificação da implantação antes de criar execuções.

Conecte um Fluxo com Permissões Restritas e Novas Tentativas Limitadas

Nesta etapa, crie uma função IAM e uma máquina de fluxo independentes que possam repetir a falha da função de processamento após a gravação.

O Step Functions precisa de confiança no serviço states e de uma concessão InvokeFunction separada para a função exata. A função de processamento mantém suas próprias permissões DynamoDB; a função IAM do fluxo apenas a invoca. Atribuições no shell salvam identificadores retornados. --query e --output text selecionam valores reutilizáveis, file:// lê o JSON literal de confiança e o shell insere o ARN da função de processamento no documento de permissões.

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

A lista de máquinas vazia confirma que nenhum fluxo de uma VM anterior é reutilizado. A concessão usa o ARN da função, enquanto cada Task usa seu nome comum de função. Retry trata apenas TransientOrderError com um intervalo de um segundo, espera progressiva que dobra e no máximo duas novas tentativas após a primeira; Catch encaminha InvalidOrder para um Fail explícito. ResultSelector mantém o Payload real, ResultPath o preserva em saved e OutputPath retorna o resumo real.

Um here-document com marcador entre aspas grava ASL literal. jq --arg substitui os marcadores da função de processamento antes da criação da 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}'

A definição nativa lista Retry/Catch seletivos e duas tarefas reais. AWS View mostra a máquina ainda sem execuções ou pedidos. Execute a verificação da configuração do fluxo.

Comprove uma Única Gravação de Negócio em Novas Tentativas e Execuções

Nesta etapa, execute uma falha após a gravação, repita o mesmo pedido e crie um pedido diferente.

Nomes de execução identificam execuções do fluxo; IDs de pedidos identificam efeitos de negócio. A primeira execução usa o modo after-write. Um laço com limite de iterações lê o status até que deixe de ser RUNNING; $(...) captura a saída e break sai do laço. Inspecione a execução se ainda estiver RUNNING após o laço.

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

O primeiro TaskFailed é TransientOrderError após a gravação real. A nova tentativa bem-sucedida retorna duplicate:true; o resumo é concluído com total 1100. O único item protected-order tem quantidade 4/total 1100. A nova tentativa da tarefa foi bem-sucedida sem um segundo PutItem aceito para essa chave de negócio.

Inicie um novo nome de execução com o mesmo ID de pedido e os mesmos valores:

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'

Essa nova execução também retorna um resultado de armazenamento duplicado e lê o resumo original de 1100. Um novo nome de execução não torna o pedido uma nova solicitação de negócio. Falhas reais da condição preservam o item original em vez de sobrescrevê-lo.

Use um ID de pedido diferente para comprovar que a proteção não rejeita trabalho não relacionado:

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'

O pedido diferente é bem-sucedido com 1/350. Há dois itens de negócio, embora tenham ocorrido sete chamadas reais à função de processamento entre tentativas de armazenamento e leituras de resumo. Logs e resultados nativos das tarefas mostram decisões de duplicata para a chave original. AWS View mostra três resumos bem-sucedidos e os dois pedidos preservados.

O exemplo abaixo mostra resumos reais de nova tentativa e de nova execução, junto com o pedido original preservado e o novo pedido separado.

AWS View mostra o pedido original preservado e um pedido de negócio diferente Esta é uma gravação idempotente para a solicitação idêntica testada, não uma promessa de que as tarefas são executadas exatamente uma vez.

Execute a verificação da proteção de negócio.

Remova os Recursos do Fluxo e os Resultados Sintéticos

Nesta etapa, exclua sua máquina concluída, a função IAM do fluxo, o pedido e os logs, preservando os recursos de apoio fornecidos.

As três execuções terminaram. Excluir a máquina a remove da lista de máquinas ativas. Remova a política da função IAM que você criou antes de excluir essa função IAM e, depois, remova o pedido sintético e o grupo de logs da função de processamento criados por sua execução.

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

Leia inventários bem-sucedidos para comprovar o que 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

Não há máquinas ativas, pedidos ou grupos de logs. Apenas a função IAM fornecida da função de processamento permanece, e o item de referência está inalterado. Mantenha a função de processamento e as tabelas fornecidas: elas foram criadas pela preparação, enquanto o fluxo e seus recursos foram criados por você. Erros de rede ou de autenticação nunca comprovam a exclusão.

Remova os arquivos comuns criados neste laboratório:

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

AWS View mostra máquinas, execuções e pedidos vazios e a referência preservada. Execute a verificação da limpeza antes de encerrar a VM.

Resumo

Você protegeu gravações nativas de negócio com uma chave estável de pedido, tratou apenas conflitos condicionais idênticos como duplicatas e testou uma falha real após a primeira gravação. Uma nova tentativa e uma nova execução do fluxo preservaram o pedido original; uma chave de negócio diferente criou seu próprio resultado. Você verificou históricos e dados nativos antes de remover os recursos do fluxo que criou e os resultados sintéticos.

Relacionados AWS Cursos
Fundamentos da AWS para iniciantes

Fundamentos da AWS para iniciantes

aws
AWS S3 para iniciantes

AWS S3 para iniciantes

aws
AWS IAM para iniciantes

AWS IAM para iniciantes

aws