Serverless DeepDive #01：Invokeの成功は何の成功か

const lambdaFunction = new lambda.Function(this, `Function-${fn.id}`, {
functionName: name,
description: `lambda-lab ${scenario.id}/${fn.id}`,
runtime: RUNTIMES[handler.runtime],
handler: handler.handler,
code:
handler.kind === "inline"
? lambda.Code.fromInline(handler.code)
: lambda.Code.fromAsset(handler.directory),
architecture:
fn.architecture === "arm64" ? lambda.Architecture.ARM_64 : lambda.Architecture.X86_64,
memorySize: fn.memoryMb,
timeout: Duration.seconds(fn.timeoutSeconds),
environment: Object.fromEntries(environment),
reservedConcurrentExecutions: fn.reservedConcurrency ?? undefined,
// Always set: without them AWS applies its own async defaults and the run is no longer bounded.
retryAttempts: fn.asyncRetryAttempts,
maxEventAge: Duration.seconds(fn.maxEventAgeSeconds),
role,
logGroup,
});

"""Compare a function response with its single S3 business result."""

import json
import os
import re
import time

import boto3
from botocore.config import Config
from botocore.exceptions import ClientError

s3 = boto3.client("s3", config=Config(retries={"total_max_attempts": 1}))

def emit(record, phase, **fields):
print(json.dumps({**record, "phase": phase, **fields}, sort_keys=True))

def handler(event, context):
if type(event) is not dict:
raise ValueError("event must be an object")
for name in ("id", "run_id"):
if type(event.get(name)) is not str or not re.fullmatch(
r"[a-z0-9-]{1,64}", event[name]
):
raise ValueError("id and run_id must match [a-z0-9-]{1,64}")
value = event.get("value")
action = event.get("action", "calculate")
policy = event.get("storage_policy", "job")
if type(value) is not int or not -1000 <= value <= 1000:
raise ValueError("value must be an integer in [-1000,1000]")
if action not in ("calculate", "fail-before", "fail-after"):
raise ValueError("unknown action")
if policy not in ("request", "job"):
raise ValueError("unknown storage_policy")
record = {
"id": event["id"],
"run_id": event["run_id"],
"request_id": context.aws_request_id,
"value": value,
"action": action,
"storage_policy": policy,
}
emit(record, "start")
if value < 0:
result = {**record, "ok": False, "reason": "negative_value"}
emit(result, "rejected")
return result
if action == "fail-before":
emit(record, "failed_before_save")
raise RuntimeError("intentional failure before saving result")
bucket = os.environ["LAB_EVIDENCE_BUCKET"]
base = os.environ["LAB_EVIDENCE_PREFIX"] + event["run_id"] + "/" + event["id"]
key = base + "/" + (
context.aws_request_id + ".json" if policy == "request" else "result.json"
)
saved = {
**record,
"ok": True,
"result": value * 2,
"commit_request_id": context.aws_request_id,
"saved_unix": time.time(),
}
try:
s3.put_object(
Bucket=bucket,
Key=key,
Body=json.dumps(saved, sort_keys=True).encode(),
ContentType="application/json",
IfNoneMatch="*",
)
except ClientError as error:
if error.response["ResponseMetadata"]["HTTPStatusCode"] != 412:
raise
response = s3.get_object(Bucket=bucket, Key=key)
try:
previous = json.loads(response["Body"].read())
finally:
response["Body"].close()
if previous["value"] != value:
result = {**record, "ok": False, "reason": "job_input_conflict"}
emit(result, "conflict")
return result
result = {
**previous,
"request_id": context.aws_request_id,
"replayed": True,
}
emit(result, "replayed")
return result
emit(saved, "saved")
if action == "fail-after":
emit(saved, "failed_after_save")
raise RuntimeError("intentional failure after saving result")
emit(saved, "returned")
return {**saved, "replayed": False}

const sync = request.invocation === "sync";
const output = await clients.lambda.send(
new InvokeCommand({
FunctionName: request.functionName,
InvocationType: sync ? "RequestResponse" : "Event",
LogType: sync ? "Tail" : undefined,
Payload: new TextEncoder().encode(request.payload),
}),
{ abortSignal },
);