test(idempotency): add e2e tests for idempotency

packages/commons/tests/utils/e2eUtils.ts

@@ -1,15 +1,12 @@
-/** 
+/**
  * E2E utils is used by e2e tests. They are helper function that calls either CDK or SDK
- * to interact with services. 
-*/
-import { App, CfnOutput, Stack, Duration } from 'aws-cdk-lib';
-import {
-  NodejsFunction,
-  NodejsFunctionProps
-} from 'aws-cdk-lib/aws-lambda-nodejs';
+ * to interact with services.
+ */
+import { App, CfnOutput, Duration, Stack } from 'aws-cdk-lib';
+import { NodejsFunction, NodejsFunctionProps } from 'aws-cdk-lib/aws-lambda-nodejs';
 import { Runtime, Tracing } from 'aws-cdk-lib/aws-lambda';
 import { RetentionDays } from 'aws-cdk-lib/aws-logs';
-import { LambdaClient, InvokeCommand } from '@aws-sdk/client-lambda';
+import { InvokeCommand, LambdaClient } from '@aws-sdk/client-lambda';
 import { fromUtf8 } from '@aws-sdk/util-utf8-node';
 
 import { InvocationLogs } from './InvocationLogs';
@@ -30,20 +27,20 @@ export type StackWithLambdaFunctionOptions = {
   functionName: string
   functionEntry: string
   tracing?: Tracing
-  environment: {[key: string]: string}
+  environment: { [key: string]: string }
   logGroupOutputKey?: string
   runtime: string
   bundling?: NodejsFunctionProps['bundling']
   layers?: NodejsFunctionProps['layers']
   timeout?: Duration
 };
 
-type FunctionPayload = {[key: string]: string | boolean | number};
+type FunctionPayload = { [key: string]: string | boolean | number | Array<Record<string, unknown>> };
 
 export const isValidRuntimeKey = (runtime: string): runtime is TestRuntimesKey => testRuntimeKeys.includes(runtime);
 
 export const createStackWithLambdaFunction = (params: StackWithLambdaFunctionOptions): Stack => {
-  
+
   const stack = new Stack(params.app, params.stackName);
   const testFunction = new NodejsFunction(stack, `testFunction`, {
     functionName: params.functionName,
@@ -62,26 +59,27 @@ export const createStackWithLambdaFunction = (params: StackWithLambdaFunctionOpt
       value: testFunction.logGroup.logGroupName,
     });
   }
-  
+
   return stack;
 };
 
-export const generateUniqueName = (name_prefix: string, uuid: string, runtime: string, testName: string): string => 
-  `${name_prefix}-${runtime}-${uuid.substring(0,5)}-${testName}`.substring(0, 64);
+export const generateUniqueName = (name_prefix: string, uuid: string, runtime: string, testName: string): string =>
+  `${name_prefix}-${runtime}-${uuid.substring(0, 5)}-${testName}`.substring(0, 64);
 
-export const invokeFunction = async (functionName: string, times: number = 1, invocationMode: 'PARALLEL' | 'SEQUENTIAL' = 'PARALLEL', payload: FunctionPayload = {}): Promise<InvocationLogs[]> => {
+export const invokeFunction = async (functionName: string, times: number = 1, invocationMode: 'PARALLEL' | 'SEQUENTIAL' = 'PARALLEL', payload: FunctionPayload = {}, includeIndex = true): Promise<InvocationLogs[]> => {
   const invocationLogs: InvocationLogs[] = [];
 
-  const promiseFactory = (index?: number): Promise<void> => {
+  const promiseFactory = (index?: number, includeIndex?: boolean): Promise<void> => {
+
+    // in some cases we need to send a payload without the index, i.e. idempotency tests
+    const payloadToSend = includeIndex ? { invocation: index, ...payload } : { ...payload };
+
     const invokePromise = lambdaClient
       .send(new InvokeCommand({
         FunctionName: functionName,
         InvocationType: 'RequestResponse',
         LogType: 'Tail', // Wait until execution completes and return all logs
-        Payload: fromUtf8(JSON.stringify({
-          invocation: index,
-          ...payload
-        })),
+        Payload: fromUtf8(JSON.stringify(payloadToSend)),
       }))
       .then((response) => {
         if (response?.LogResult) {
@@ -93,17 +91,17 @@ export const invokeFunction = async (functionName: string, times: number = 1, in
 
     return invokePromise;
   };
-  
-  const promiseFactories = Array.from({ length: times }, () => promiseFactory );
+
+  const promiseFactories = Array.from({ length: times }, () => promiseFactory);
   const invocation = invocationMode == 'PARALLEL'
-    ? Promise.all(promiseFactories.map((factory, index) => factory(index)))
+    ? Promise.all(promiseFactories.map((factory, index) => factory(index, includeIndex)))
     : chainPromises(promiseFactories);
   await invocation;
 
-  return invocationLogs; 
+  return invocationLogs;
 };
 
-const chainPromises = async (promiseFactories: ((index?: number) => Promise<void>)[]) : Promise<void> => {
+const chainPromises = async (promiseFactories: ((index?: number) => Promise<void>)[]): Promise<void> => {
   for (let index = 0; index < promiseFactories.length; index++) {
     await promiseFactories[index](index);
   }

packages/idempotency/package.json

@@ -13,10 +13,10 @@
     "commit": "commit",
     "test": "npm run test:unit",
     "test:unit": "jest --group=unit --detectOpenHandles --coverage --verbose",
-    "test:e2e:nodejs14x": "echo \"Not implemented\"",
-    "test:e2e:nodejs16x": "echo \"Not implemented\"",
-    "test:e2e:nodejs18x": "echo \"Not implemented\"",
-    "test:e2e": "echo \"Not implemented\"",
+    "test:e2e:nodejs14x": "RUNTIME=nodejs14x jest --group=e2e",
+    "test:e2e:nodejs16x": "RUNTIME=nodejs16x jest --group=e2e",
+    "test:e2e:nodejs18x": "RUNTIME=nodejs18x jest --group=e2e",
+    "test:e2e": "jest --group=e2e --detectOpenHandles",
     "watch": "jest --watch --group=unit",
     "build": "tsc",
     "lint": "eslint --ext .ts --no-error-on-unmatched-pattern src tests",
@@ -57,6 +57,7 @@
   ],
   "devDependencies": {
     "@types/jmespath": "^0.15.0",
+    "@aws-sdk/client-dynamodb": "^3.231.0",
     "aws-sdk-client-mock": "^2.0.1",
     "aws-sdk-client-mock-jest": "^2.0.1"
   }

packages/idempotency/src/IdempotencyHandler.ts

@@ -1,25 +1,25 @@
-import type { AnyFunctionWithRecord, IdempotencyOptions } from './types';
+import type { AnyFunctionWithRecord } from './types';
 import { IdempotencyRecordStatus } from './types';
 import {
   IdempotencyAlreadyInProgressError,
   IdempotencyInconsistentStateError,
   IdempotencyItemAlreadyExistsError,
   IdempotencyPersistenceLayerError,
 } from './Exceptions';
-import { IdempotencyRecord } from './persistence';
+import { BasePersistenceLayer, IdempotencyRecord } from './persistence';
+import { IdempotencyConfig } from 'IdempotencyConfig';
 
 export class IdempotencyHandler<U> {
   public constructor(
     private functionToMakeIdempotent: AnyFunctionWithRecord<U>,
     private functionPayloadToBeHashed: Record<string, unknown>,
-    private idempotencyOptions: IdempotencyOptions,
+    private config: IdempotencyConfig,
+    private persistenceStore: BasePersistenceLayer,
     private fullFunctionPayload: Record<string, unknown>,
   ) {
   }
 
-  public determineResultFromIdempotencyRecord(
-    idempotencyRecord: IdempotencyRecord
-  ): Promise<U> | U {
+  public determineResultFromIdempotencyRecord(idempotencyRecord: IdempotencyRecord): Promise<U> | U {
     if (idempotencyRecord.getStatus() === IdempotencyRecordStatus.EXPIRED) {
       throw new IdempotencyInconsistentStateError(
         'Item has expired during processing and may not longer be valid.'
@@ -40,10 +40,31 @@ export class IdempotencyHandler<U> {
           `There is already an execution in progress with idempotency key: ${idempotencyRecord.idempotencyKey}`
         );
       }
-    } else {
-      // Currently recalling the method as this fulfills FR1. FR3 will address using the previously stored value https://github.com/awslabs/aws-lambda-powertools-typescript/issues/447
-      return this.functionToMakeIdempotent(this.fullFunctionPayload);
     }
+
+    return idempotencyRecord.getResponse() as U;
+  }
+
+  public async getFunctionResult(): Promise<U> {
+    let result: U;
+    try {
+      result = await this.functionToMakeIdempotent(this.fullFunctionPayload);
+
+    } catch (e) {
+      try {
+        await this.persistenceStore.deleteRecord(this.functionPayloadToBeHashed);
+      } catch (e) {
+        throw new IdempotencyPersistenceLayerError('Failed to delete record from idempotency store');
+      }
+      throw e;
+    }
+    try {
+      await this.persistenceStore.saveSuccess(this.functionPayloadToBeHashed, result as Record<string, unknown>);
+    } catch (e) {
+      throw new IdempotencyPersistenceLayerError('Failed to update success record to idempotency store');
+    }
+
+    return result;
   }
 
   /**
@@ -70,13 +91,13 @@ export class IdempotencyHandler<U> {
 
   public async processIdempotency(): Promise<U> {
     try {
-      await this.idempotencyOptions.persistenceStore.saveInProgress(
+      await this.persistenceStore.saveInProgress(
         this.functionPayloadToBeHashed,
       );
     } catch (e) {
       if (e instanceof IdempotencyItemAlreadyExistsError) {
         const idempotencyRecord: IdempotencyRecord =
-          await this.idempotencyOptions.persistenceStore.getRecord(
+          await this.persistenceStore.getRecord(
             this.functionPayloadToBeHashed
           );
 
@@ -86,6 +107,7 @@ export class IdempotencyHandler<U> {
       }
     }
 
-    return this.functionToMakeIdempotent(this.fullFunctionPayload);
+    return this.getFunctionResult();
   }
+
 }

packages/idempotency/src/idempotentDecorator.ts

@@ -1,15 +1,22 @@
-import {
-  GenericTempRecord,
-  IdempotencyOptions,
-} from './types';
+import { GenericTempRecord, IdempotentOptions, } from './types';
 import { IdempotencyHandler } from './IdempotencyHandler';
+import { IdempotencyConfig } from './IdempotencyConfig';
 
-const idempotent = function (options: IdempotencyOptions) {
+const idempotent = function (options: IdempotentOptions) {
   return function (_target: unknown, _propertyKey: string, descriptor: PropertyDescriptor) {
     const childFunction = descriptor.value;
     // TODO: sort out the type for this
-    descriptor.value = function(record: GenericTempRecord){
-      const idempotencyHandler = new IdempotencyHandler<GenericTempRecord>(childFunction, record[options.dataKeywordArgument], options, record);
+
+    descriptor.value = function (record: GenericTempRecord) {
+      const config = options.config || new IdempotencyConfig({});
+      const dataKeywordArgument = options?.dataKeywordArgument ? record[options?.dataKeywordArgument] : record;
+      config.registerLambdaContext(record.context);
+      const idempotencyHandler = new IdempotencyHandler<GenericTempRecord>(
+        childFunction,
+        dataKeywordArgument,
+        config,
+        options.persistenceStore,
+        record);
 
       return idempotencyHandler.handle();
     };

packages/idempotency/src/makeFunctionIdempotent.ts

@@ -1,17 +1,22 @@
-import type {
-  GenericTempRecord,
-  IdempotencyOptions,
-  AnyFunctionWithRecord,
-  AnyIdempotentFunction,
-} from './types';
+import type { AnyFunctionWithRecord, AnyIdempotentFunction, GenericTempRecord, IdempotentOptions, } from './types';
 import { IdempotencyHandler } from './IdempotencyHandler';
+import { IdempotencyConfig } from './IdempotencyConfig';
 
 const makeFunctionIdempotent = function <U>(
   fn: AnyFunctionWithRecord<U>,
-  options: IdempotencyOptions
+  options: IdempotentOptions,
 ): AnyIdempotentFunction<U> {
   const wrappedFn: AnyIdempotentFunction<U> = function (record: GenericTempRecord): Promise<U> {
-    const idempotencyHandler: IdempotencyHandler<U> = new IdempotencyHandler<U>(fn, record[options.dataKeywordArgument], options, record);
+    if (options.dataKeywordArgument === undefined) {
+      throw new Error(`Missing data keyword argument ${options.dataKeywordArgument}`);
+    }
+    const config = new IdempotencyConfig({});
+    const idempotencyHandler: IdempotencyHandler<U> = new IdempotencyHandler<U>(
+      fn,
+      record[options.dataKeywordArgument],
+      config,
+      options.persistenceStore,
+      record);
 
     return idempotencyHandler.handle();
   };

packages/idempotency/src/types/IdempotencyOptions.ts

@@ -1,9 +1,11 @@
 import type { Context } from 'aws-lambda';
 import { BasePersistenceLayer } from '../persistence/BasePersistenceLayer';
+import { IdempotencyConfig } from 'IdempotencyConfig';
 
-type IdempotencyOptions = {
-  dataKeywordArgument: string
+type IdempotentOptions = {
   persistenceStore: BasePersistenceLayer
+  dataKeywordArgument?: string
+  config?: IdempotencyConfig
 };
 
 /**
@@ -45,6 +47,6 @@ type IdempotencyConfigOptions = {
 };
 
 export {
-  IdempotencyOptions,
-  IdempotencyConfigOptions
+  IdempotencyConfigOptions,
+  IdempotentOptions
 };

packages/idempotency/tests/e2e/constants.ts

@@ -0,0 +1,6 @@
+export const RESOURCE_NAME_PREFIX = 'Idempotency-E2E';
+
+export const ONE_MINUTE = 60 * 1_000;
+export const TEARDOWN_TIMEOUT = 5 * ONE_MINUTE;
+export const SETUP_TIMEOUT = 5 * ONE_MINUTE;
+export const TEST_CASE_TIMEOUT = 5 * ONE_MINUTE;