Les traductions sont fournies par des outils de traduction automatique. En cas de conflit entre le contenu d'une traduction et celui de la version originale en anglais, la version anglaise prévaudra.

Mappages de sources d'événements avec fonctions durables

Les fonctions durables fonctionnent avec tous les mappages de sources d'événements Lambda. Configurez les mappages de sources d'événements pour les fonctions durables de la même manière que vous les configurez pour les fonctions standard. Les mappages de sources d'événements interrogent automatiquement des sources d'événements telles qu'Amazon SQS, Kinesis et DynamoDB Streams, et invoquent votre fonction avec des lots d'enregistrements.

Les mappages de sources d'événements sont utiles pour les fonctions durables qui traitent des flux ou des files d'attente avec des flux de travail complexes en plusieurs étapes. Par exemple, vous pouvez créer une fonction durable qui traite les messages Amazon SQS avec de nouvelles tentatives, des appels d'API externes et des approbations humaines.

Comment les mappages de sources d'événements invoquent des fonctions durables

Les mappages de sources d'événements invoquent des fonctions durables de manière synchrone, en attendant la fin de l'exécution durable complète avant de traiter le lot suivant ou de marquer les enregistrements comme traités. Si le temps d'exécution durable total dépasse 15 minutes, l'exécution expire et échoue. Le mappage des sources d'événements reçoit une exception de délai d'attente et la gère en fonction de sa configuration de nouvelle tentative.

Limite d'exécution de 15 minutes

Lorsque des fonctions durables sont invoquées par des mappages de sources d'événements, la durée totale d'exécution durable ne peut pas dépasser 15 minutes. Cette limite s'applique à l'ensemble de l'exécution durable du début à la fin, et pas uniquement aux appels de fonctions individuels.

Cette limite de 15 minutes est distincte du délai d'expiration de la fonction Lambda (également 15 minutes maximum). Le délai d'expiration de la fonction contrôle la durée pendant laquelle chaque appel individuel peut s'exécuter, tandis que le délai d'exécution durable contrôle le temps total écoulé entre le début et la fin de l'exécution.

Exemples de scénarios :

Configuration des mappages de sources d'événements

Configurez des mappages de sources d'événements pour des fonctions durables à l'aide de la console Lambda ou AWS des AWS CLI kits SDK. Toutes les propriétés de mappage de sources d'événements standard s'appliquent aux fonctions durables :

aws lambda create-event-source-mapping \
  --function-name arn:aws:lambda:us-east-1:123456789012:function:my-durable-function:1 \
  --event-source-arn arn:aws:sqs:us-east-1:123456789012:my-queue \
  --batch-size 10 \
  --maximum-batching-window-in-seconds 5
    

N'oubliez pas d'utiliser un ARN qualifié (avec numéro de version ou alias) lors de la configuration des mappages de sources d'événements pour les fonctions durables.

Gestion des erreurs avec les mappages de sources d'événements

Les mappages de sources d'événements fournissent une gestion des erreurs intégrée qui fonctionne avec des fonctions durables :

Pour des informations complètes sur la gestion des erreurs de mappage des sources d'événements, consultez la section Mappages des sources d'événements.

Utilisation d'une fonction intermédiaire pour les flux de travail de longue durée

Si votre flux de travail nécessite plus de 15 minutes, utilisez une fonction Lambda standard intermédiaire entre le mappage des sources d'événements et votre fonction durable. La fonction intermédiaire reçoit les événements du mappage des sources d'événements et invoque la fonction durable de manière asynchrone, supprimant ainsi la limite d'exécution de 15 minutes.

Ce modèle découple le modèle d'appel synchrone du mappage des sources d'événements du modèle d'exécution de longue durée de la fonction durable. Le mappage des sources d'événements invoque la fonction intermédiaire, qui revient rapidement après le démarrage de l'exécution durable. La fonction durable fonctionne ensuite de manière autonome aussi longtemps que nécessaire (jusqu'à 1 an).

Architecture

Le modèle de fonction intermédiaire utilise trois composants :

Mise en œuvre

La fonction intermédiaire reçoit l'intégralité de l'événement à partir du mappage des sources d'événements et invoque la fonction durable de manière asynchrone. Utilisez le paramètre du nom d'exécution pour vous assurer que l'exécution idempotente démarre, évitant ainsi le traitement dupliqué si le mappage de la source de l'événement tente à nouveau :

TypeScript

import { LambdaClient, InvokeCommand } from '@aws-sdk/client-lambda';
import { SQSEvent } from 'aws-lambda';
import { createHash } from 'crypto';

const lambda = new LambdaClient({});

export const handler = async (event: SQSEvent) => {
  // Invoke durable function asynchronously with execution name
  await lambda.send(new InvokeCommand({
    FunctionName: 'arn:aws:lambda:us-east-1:123456789012:function:my-durable-function:1',
    InvocationType: 'Event',
    Payload: JSON.stringify({
      executionName: event.Name,
      event: event
    })
  }));
  
  return { statusCode: 200 };
};
            

Python

import boto3
import json
import hashlib

lambda_client = boto3.client('lambda')

def handler(event, context):  
    # Invoke durable function asynchronously with execution name
    lambda_client.invoke(
        FunctionName='arn:aws:lambda:us-east-1:123456789012:function:my-durable-function:1',
        InvocationType='Event',
        Payload=json.dumps({
            'executionName': execution_name,
            'event': event["name"]
        })
    )
    
    return {'statusCode': 200}
            

Pour l'idempotence dans la fonction intermédiaire elle-même, utilisez Powertools pour éviter les invocations dupliquées AWS Lambda de la fonction durable si le mappage de la source d'événements tente à nouveau la fonction intermédiaire.

La fonction durable reçoit la charge utile avec le nom d'exécution et traite tous les enregistrements avec une logique de longue durée :

TypeScript

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

export const handler = withDurableExecution(
  async (payload: any, context: DurableContext) => {
    const sqsEvent = payload.event;
    
    // Process each record with complex, multi-step logic
    const results = await context.map(
      sqsEvent.Records,
      async (ctx, record) => {
        const validated = await ctx.step('validate', async () => {
          return validateOrder(JSON.parse(record.body));
        });
        
        // Wait for external approval (could take hours or days)
        const approval = await ctx.waitForCallback(
          'approval',
          async (callbackId) => {
            await requestApproval(callbackId, validated);
          },
          { timeout: { hours: 48 } }
        );
        
        // Complete processing
        return await ctx.step('complete', async () => {
          return completeOrder(validated, approval);
        });
      }
    );
    
    return { statusCode: 200, processed: results.getResults().length };
  }
);
            

Python

from aws_durable_execution_sdk_python import durable_execution, DurableContext
from aws_durable_execution_sdk_python.config import Duration, WaitForCallbackConfig
from collections.abc import Sequence
import json

def validate_order(order_data: dict) -> dict:
    """Validate order data - always passes."""
    return order_data

def request_approval(callback_id: str, validated_order: dict) -> None:
    """Request approval for the order - always passes."""
    pass

def complete_order(validated_order: dict, approval_result: str) -> dict:
    """Complete the order processing - always passes."""
    return validated_order

@durable_execution
def lambda_handler(payload, context: DurableContext):
    sqs_event = payload['event']

    def process_record(
        ctx: DurableContext, 
        record: dict, 
        index: int, 
        items: Sequence[dict]
    ) -> dict:
        validated = ctx.step(
            lambda _: validate_order(json.loads(record['body'])),
            name=f'validate-{index}'
        )

        approval = ctx.wait_for_callback(
            submitter=lambda callback_id, wait_ctx: request_approval(callback_id, validated),
            name=f'approval-{index}',
            config=WaitForCallbackConfig(timeout=Duration.from_seconds(172800))
        )

        return ctx.step(
            lambda _: complete_order(validated, approval),
            name=f'complete-{index}'
        )

    results = context.map(
        inputs=sqs_event['Records'],
        func=process_record,
        name='process-records'
    )

    return {
        'statusCode': 200, 
        'started': results.started_count,
        'completed': results.success_count,
        'failed': results.failure_count,
        'total': results.total_count
    }
            

Considérations clés

Ce modèle supprime la limite d'exécution de 15 minutes en découplant le mappage de la source d'événements de l'exécution durable. La fonction intermédiaire revient immédiatement après le démarrage de l'exécution durable, ce qui permet au mappage de la source d'événements de poursuivre le traitement. La fonction durable fonctionne ensuite de manière autonome aussi longtemps que nécessaire.

La fonction intermédiaire réussit lorsqu'elle invoque la fonction durable, et non lorsque l'exécution durable est terminée. Si l'exécution durable échoue ultérieurement, le mappage de la source d'événements ne sera pas réessayé car il a déjà traité le lot avec succès. Implémentez la gestion des erreurs dans la fonction durable et configurez des files d'attente de lettres mortes pour les échecs d'exécution.

Utilisez le paramètre du nom d'exécution pour vous assurer que l'exécution idempotente démarre. Si le mappage des sources d'événements réessaie la fonction intermédiaire, la fonction durable ne lancera pas d'exécution dupliquée car le nom d'exécution existe déjà.

Sources d'événements prises en charge

Les fonctions durables prennent en charge toutes les sources d'événements Lambda qui utilisent des mappages de sources d'événements :

Tous les types de sources d'événements sont soumis à la limite d'exécution durable de 15 minutes lors de l'appel de fonctions durables.