Wiederholte Geschäftseffekte durch Workflow-Wiederholungen verhindern

Einführung

Ein Worker speichert eine Bestellung, schlägt aber fehl, bevor sein Ergebnis den Workflow erreicht. Du schützt diesen Schreibvorgang, damit eine Wiederholung oder erneute Ausführung die ursprüngliche Bestellung erhält, während eine andere Bestellung weiterhin erfolgreich sein kann.

Step Functions koordiniert Lambda-Aufgaben; der bereitgestellte Worker speichert Bestellungen in DynamoDB und liest ihre Zusammenfassungen. Der CLI-Zugriff ist bereits eingerichtet; ein persönliches AWS-Konto oder Zugriffsschlüssel sind nicht erforderlich.

Schließe zuerst Einen fehlgeschlagenen Schritt wiederholen und dauerhafte Fehler behandeln, Doppelte Bestellungen mit bedingten Schreibvorgängen verhindern und Eine Lambda-Funktion konfigurieren und diagnostizieren ab. Diese unabhängige VM stellt einen ungeschützten Worker bereit; du ergänzt den Schutz.

Bezug zu Zertifizierungen

Laborübersicht

Den Geschäftsschreibvorgang mit einem stabilen Schlüssel schützen

Stelle in diesem Schritt einen Worker bereit, der eine wiederholte identische Bestellung als bereits abgeschlossenen Schreibvorgang behandelt.

Verwende AWS View oben, um die CLI-Abfragen mit den tatsächlichen Ressourcen und Ergebnissen dieses Labs zu vergleichen. Erhalte die bereitgestellten Referenzdaten.

Die Bestell-ID ist ein Geschäftsschlüssel: Sie bezeichnet die vorgesehene Bestellung, auch wenn Aufgabenversuche oder Workflow-Ausführungsnamen unterschiedlich sind. attribute_not_exists(id) erlaubt nur das erste PutItem. DynamoDB wertet diese Bedingung atomar aus. ReturnValuesOnConditionCheckFailure:ALL_OLD liefert das vorhandene Element, wenn eine Wiederholung beim Prüfen der Bedingung unterliegt. Gib nur dann ein Duplikatergebnis zurück, wenn dieses vorhandene Element der vorgesehenen Bestellung entspricht; widersprüchlicher Inhalt muss weiterhin fehlschlagen, statt eine Bestellung stillschweigend zu ersetzen.

Der bereitgestellte Ausgangscode hat einen bedingungslosen Schreibvorgang. Ersetze ihn durch den vollständigen geschützten Handler unten. Er verwendet die konfigurierte SDK-Umgebung, keine persönlichen Zugangsdaten. Diagnoseversuche verfolgen tatsächliche Aufrufe getrennt von geschäftlichen Schreibvorgängen. Im fiktiven Modus after-write löst der erste Versuch nach dem Schreibvorgang beim Dienst einen Fehler aus und erzeugt so die Unsicherheit, mit der eine Wiederholung umgehen muss. Der Modus summary liest das tatsächlich gespeicherte Element.

Ein Here-Dokument mit einer in Anführungszeichen gesetzten Markierung schreibt die wörtliche Python-Datei. Der vorhandene Lambda-Handler bleibt app.handler.

Der Schutz gehört an den Geschäftsschreibvorgang, weil ein Ausführungsname keine Bestellung schützt, auf die mehrere Ausführungen zugreifen. Nur ein identisches vorhandenes Element gilt als erledigte Arbeit. Dieselbe Bestell-ID mit anderem Inhalt ist ein zu meldender Konflikt und keine Erlaubnis zum Überschreiben.

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

Das try führt den bedingten Schreibvorgang beim Dienst aus. Nur eine tatsächliche ConditionalCheckFailedException mit einem identischen alten Element wird als Duplikat behandelt; andere SDK-Fehler werden erneut ausgelöst. Die temporäre Ausnahme tritt nach dieser Schreibentscheidung auf, sodass der Workflow tatsächlich Arbeit wiederholt, deren Geschäftselement bereits vorhanden ist.

Verpacke die Datei mit zip -j, das Verzeichnispfade weglässt, und aktualisiere den bereitgestellten Worker mit --zip-file fileb:// für binäre Archivbytes. CodeSha256 in der Antwort bezeichnet das bereitgestellte Archiv. Die Bereitstellung allein belegt keinen Duplikatschutz; der Laufzeitschritt wird ihn testen.

zip -j function.zip app.py

Wende das neue Funktionscodepaket an:

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

Führe die Bereitstellungsprüfung aus, bevor du Ausführungen erstellst.

Einen begrenzten Workflow mit begrenzten Wiederholungen verbinden

Erstelle in diesem Schritt eine unabhängige Workflow-Rolle und Maschine, die den Fehler des Workers nach dem Schreiben wiederholen kann.

Step Functions benötigt Vertrauen für den states-Dienst und eine separate InvokeFunction-Berechtigung für genau die Funktion. Der Worker behält seine eigenen DynamoDB-Berechtigungen; die Workflow-Rolle ruft ihn nur auf. Shellzuweisungen speichern zurückgegebene Kennungen. --query und --output text wählen wiederverwendbare Werte aus, file:// liest das wörtliche Vertrauens-JSON, und die Shell setzt die Worker-ARN in das Berechtigungsdokument ein.

cd /home/labex/project
WORKER_NAME=labex-ev05-worker

Speichere den zurückgegebenen Wert in WORKER_ARN:

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

Prüfe die Rolle und Umgebungskonfiguration der Funktion:

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

Prüfe die verbleibenden Zustandsmaschinen:

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

Speichere den zurückgegebenen Wert in 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

Wende die Berechtigungsrichtlinie der Anwendungsrolle an:

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

Prüfe die Berechtigungsrichtlinie der Anwendungsrolle:

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

Die leere Maschinenliste bestätigt, dass kein Workflow einer früheren VM wiederverwendet wird. Die Berechtigung verwendet die Funktions-ARN, während jede Task ihren gewöhnlichen Funktionsnamen verwendet. Retry behandelt nur TransientOrderError mit einem Intervall von einer Sekunde, verdoppelter Wartezeit und höchstens zwei Wiederholungen nach dem ersten Versuch; Catch leitet InvalidOrder zu einem ausdrücklichen Fail weiter. ResultSelector behält die tatsächliche Payload bei, ResultPath erhält sie unter saved, und OutputPath gibt die tatsächliche Zusammenfassung zurück.

Ein Here-Dokument mit einer in Anführungszeichen gesetzten Markierung schreibt wörtliches ASL. jq --arg ersetzt die Worker-Platzhalter vor der Maschinenerstellung.

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

Schreibe die Konfiguration in workflow.json:

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

Speichere den zurückgegebenen Wert in 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"

Prüfe die Zustandsmaschinendefinition und Rolle:

aws stepfunctions describe-state-machine \
  --state-machine-arn "$MACHINE_ARN" \
  --query '{Name:name,Definition:definition}'

Die Definition beim Dienst listet die gezielten Retry/Catch-Blöcke und zwei tatsächliche Aufgaben auf. AWS View zeigt die Maschine noch ohne Ausführungen oder Bestellungen. Führe die Workflow-Konfigurationsprüfung aus.

Einen Geschäftsschreibvorgang über Wiederholung und erneute Ausführung hinweg nachweisen

Führe in diesem Schritt einen Fehler nach dem Schreiben aus, wiederhole dieselbe Bestellung und erstelle eine andere Bestellung.

Ausführungsnamen bezeichnen Workflow-Durchläufe; Bestell-IDs bezeichnen Geschäftseffekte. Der erste Durchlauf verwendet den Modus after-write. Eine begrenzte Schleife liest den Status, bis er nicht mehr RUNNING ist; $(...) erfasst die Ausgabe, und break verlässt die Schleife. Untersuche die Ausführung, wenn sie nach der Schleife weiterhin RUNNING ist.

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"

Frage diese Ausführung ab, bis sie RUNNING verlässt:

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

Prüfe den Ausführungsstatus, die Ausgabe und Fehlerdetails:

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

Prüfe den tatsächlichen Workflow-Ausführungsverlauf:

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

Prüfe den tatsächlich gespeicherten Datenbankeintrag:

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

Das erste TaskFailed ist TransientOrderError nach dem tatsächlichen Schreibvorgang. Die erfolgreiche Wiederholung gibt duplicate:true zurück; die Zusammenfassung wird mit dem Gesamtbetrag 1100 abgeschlossen. Das einzige protected-order-Element hat die Menge 4 und den Gesamtbetrag 1100. Die Aufgabenwiederholung war erfolgreich, ohne ein zweites akzeptiertes PutItem für diesen Geschäftsschlüssel.

Starte einen neuen Ausführungsnamen mit derselben Bestell-ID und denselben Werten:

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"

Frage diese Ausführung ab, bis sie RUNNING verlässt:

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

Prüfe den Ausführungsstatus, die Ausgabe und Fehlerdetails:

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

Prüfe den tatsächlichen Workflow-Ausführungsverlauf:

aws stepfunctions get-execution-history \
  --execution-arn "$REPEAT_ARN" \
  --query 'events[?type==`TaskSucceeded`].taskSucceededEventDetails.output'

Auch dieser neue Durchlauf liefert ein Duplikatergebnis beim Speichern und liest die ursprüngliche Zusammenfassung mit 1100. Ein neuer Ausführungsname macht die Bestellung nicht zu einer neuen geschäftlichen Anfrage. Tatsächliche Bedingungsfehler erhalten das ursprüngliche Element, statt es zu überschreiben.

Verwende eine andere Bestell-ID, um nachzuweisen, dass der Schutz unbeteiligte Arbeit nicht zurückweist:

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"

Frage diese Ausführung ab, bis sie RUNNING verlässt:

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

Prüfe den Ausführungsstatus, die Ausgabe und Fehlerdetails:

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

Prüfe die tatsächlich gespeicherten Bestellungen:

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

Prüfe die tatsächlichen Ausführungsprotokollmeldungen:

aws logs filter-log-events \
  --log-group-name /aws/lambda/labex-ev05-worker \
  --query 'events[].message'

Die andere Bestellung ist mit 1/350 erfolgreich. Es gibt zwei Geschäftselemente, obwohl über Speicherversuche und Zusammenfassungsabfragen hinweg sieben tatsächliche Worker-Aufrufe stattgefunden haben. Logs und Aufgabenergebnisse beim Dienst zeigen Duplikatentscheidungen für den ursprünglichen Schlüssel. AWS View zeigt drei erfolgreiche Zusammenfassungen und beide erhaltenen Bestellungen.

Dies ist ein idempotenter Schreibvorgang für die getestete identische Anfrage, kein Versprechen, dass Aufgaben genau einmal ausgeführt werden.

Führe die Prüfung des Geschäftsschutzes aus.

Workflow-Ressourcen und fiktive Ergebnisse entfernen

Lösche in diesem Schritt deine abgeschlossene Maschine, Workflow-Rolle, Bestellung und Logs und erhalte dabei die vorbereiteten Ressourcen.

Alle drei Ausführungen sind beendet. Das Löschen der Maschine entfernt sie aus der Liste aktiver Maschinen. Entferne die eigene Rollenrichtlinie, bevor du die Rolle löschst, und entferne dann die fiktive Bestellung und die Worker-Log-Gruppe, die durch deine Ausführung erstellt wurden.

aws stepfunctions delete-state-machine --state-machine-arn "$MACHINE_ARN"

Lösche nur die Berechtigungsrichtlinie der Anwendungsrolle:

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

Lösche nur die Workflow-Ausführungsrolle der Übung:

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

Lösche nur den tatsächlich gespeicherten Datenbankeintrag:

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

Lösche nur den tatsächlich gespeicherten Datenbankeintrag:

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

Lösche nur den tatsächlich gespeicherten Datenbankeintrag:

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

Lösche nur den tatsächlich gespeicherten Datenbankeintrag:

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

Lösche nur die angegebene Anwendungsprotokollgruppe:

aws logs delete-log-group --log-group-name /aws/lambda/labex-ev05-worker

Lies erfolgreiche Bestandsabfragen, um zu belegen, was erhalten bleibt:

aws stepfunctions list-state-machines

Prüfe die verbleibenden IAM-Rollen:

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

Prüfe die tatsächlich gespeicherten Bestellungen:

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

Prüfe die verbleibenden Protokollgruppen:

aws logs describe-log-groups --query logGroups

Prüfe den nicht betroffenen Referenzdatensatz:

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

Es gibt keine aktiven Maschinen, Bestellungen oder Log-Gruppen. Nur die bereitgestellte Worker-Rolle bleibt erhalten, und das Referenzelement ist unverändert. Behalte den bereitgestellten Worker und die Tabellen bei: Ihre Zugehörigkeit zum Setup unterscheidet sich von deinem erstellten Workflow und deinen Ressourcen. Netzwerk- oder Authentifizierungsfehler belegen niemals eine Löschung.

Entferne die gewöhnlichen Dateien, die in diesem Lab erstellt wurden:

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

AWS View zeigt leere Maschinen-, Ausführungs- und Bestelllisten und die erhaltene Referenz. Führe die Bereinigungsprüfung aus, bevor du die VM beendest.

Zusammenfassung

Du hast Geschäftsschreibvorgänge beim Dienst mit einem stabilen Bestellschlüssel geschützt, nur identische bedingte Konflikte als Duplikate behandelt und einen tatsächlichen Fehler nach dem ersten Schreiben getestet. Wiederholung und eine neue Workflow-Ausführung erhielten die ursprüngliche Bestellung; ein anderer Geschäftsschlüssel erzeugte sein eigenes Ergebnis. Du hast Verläufe und Daten beim Dienst geprüft, bevor du eigene Workflow-Ressourcen und fiktive Ergebnisse entfernt hast.

Weitere Informationen bietet die AWS-Dokumentation: Bedingte DynamoDB-Schreibvorgänge und Überschreibschutz; Retry, Backoff und Catch-Fehlerbehandlung.