diff --git a/conformance-tests/src/main/java/plugin/ConformanceLoggingPlugin.java b/conformance-tests/src/main/java/plugin/ConformanceLoggingPlugin.java index c5444d105..30da516f3 100644 --- a/conformance-tests/src/main/java/plugin/ConformanceLoggingPlugin.java +++ b/conformance-tests/src/main/java/plugin/ConformanceLoggingPlugin.java @@ -83,7 +83,7 @@ public void onUserFunctionStart(UserFunctionStartInfo info) { @Override public void onUserFunctionEnd(UserFunctionEndInfo info) { if (isStep(info.type()) && info.attempt() != null) { - String outcome = info.succeeded() ? "SUCCEEDED" : "FAILED"; + String outcome = info.outcome().name(); System.out.println(String.format( "{\"plugin\": \"%s\", \"hook\": \"attempt-end\", \"n\": %d, \"outcome\": \"%s\", \"op\": \"%s\"%s}", prefix, info.attempt(), outcome, info.id(), arnField())); diff --git a/conformance-tests/src/main/java/plugin/PluginFaultyAndHealthy.java b/conformance-tests/src/main/java/plugin/PluginFaultyAndHealthy.java index 71760a132..9393f5527 100644 --- a/conformance-tests/src/main/java/plugin/PluginFaultyAndHealthy.java +++ b/conformance-tests/src/main/java/plugin/PluginFaultyAndHealthy.java @@ -141,7 +141,7 @@ public void onUserFunctionEnd(UserFunctionEndInfo info) { System.out.println(String.format( "{\"plugin\": \"CONFPLUGIN-HEALTHY\", \"hook\": \"attempt-end\", \"op\": \"%s\", " + "\"outcome\": \"%s\"%s}", - info.id(), info.succeeded() ? "SUCCEEDED" : "FAILED", PluginSupport.arnField(executionArn))); + info.id(), info.outcome().name(), PluginSupport.arnField(executionArn))); } @Override diff --git a/conformance-tests/src/main/java/plugin/PluginParallelBranchHooks.java b/conformance-tests/src/main/java/plugin/PluginParallelBranchHooks.java index 19a5db339..ec1e949c7 100644 --- a/conformance-tests/src/main/java/plugin/PluginParallelBranchHooks.java +++ b/conformance-tests/src/main/java/plugin/PluginParallelBranchHooks.java @@ -65,7 +65,7 @@ public void onUserFunctionEnd(UserFunctionEndInfo info) { if (!PluginSupport.isBranch(info.subType())) { return; } - String outcome = info.succeeded() ? "SUCCEEDED" : "FAILED"; + String outcome = info.outcome().name(); System.out.println(String.format( "{\"plugin\": \"CONFPLUGIN\", \"hook\": \"fn-end\", \"op\": \"%s\", \"parent\": \"%s\", " + "\"outcome\": \"%s\"%s}", diff --git a/conformance-tests/src/main/java/plugin/PluginRetryExhaustion.java b/conformance-tests/src/main/java/plugin/PluginRetryExhaustion.java index 4d0a05347..2f334f06c 100644 --- a/conformance-tests/src/main/java/plugin/PluginRetryExhaustion.java +++ b/conformance-tests/src/main/java/plugin/PluginRetryExhaustion.java @@ -67,7 +67,7 @@ public void onUserFunctionEnd(UserFunctionEndInfo info) { if (!PluginSupport.isStep(info.type()) || info.attempt() == null) { return; } - String outcome = info.succeeded() ? "SUCCEEDED" : "FAILED"; + String outcome = info.outcome().name(); System.out.println(String.format( "{\"plugin\": \"CONFPLUGIN\", \"hook\": \"attempt-end\", \"n\": %d, \"outcome\": \"%s\", " + "\"op\": \"%s\"%s}", diff --git a/examples/src/main/java/software/amazon/lambda/durable/examples/general/PluginExample.java b/examples/src/main/java/software/amazon/lambda/durable/examples/general/PluginExample.java index 85cb49bc3..1de876701 100644 --- a/examples/src/main/java/software/amazon/lambda/durable/examples/general/PluginExample.java +++ b/examples/src/main/java/software/amazon/lambda/durable/examples/general/PluginExample.java @@ -20,11 +20,11 @@ * [PLUGIN] onInvocationStart: requestId=..., durableExecutionArn=..., firstInvocation=true * [PLUGIN] onOperationStart: name=create-greeting, type=STEP * [PLUGIN] onUserFunctionStart: name=create-greeting, attempt=1 - * [PLUGIN] onUserFunctionEnd: name=create-greeting, succeeded=true + * [PLUGIN] onUserFunctionEnd: name=create-greeting, outcome=SUCCEEDED * [PLUGIN] onOperationEnd: name=create-greeting * [PLUGIN] onOperationStart: name=transform, type=STEP * [PLUGIN] onUserFunctionStart: name=transform, attempt=1 - * [PLUGIN] onUserFunctionEnd: name=transform, succeeded=true + * [PLUGIN] onUserFunctionEnd: name=transform, outcome=SUCCEEDED * [PLUGIN] onOperationEnd: name=transform * [PLUGIN] onInvocationEnd: status=SUCCEEDED * @@ -91,9 +91,9 @@ public void onUserFunctionStart(UserFunctionStartInfo info) { @Override public void onUserFunctionEnd(UserFunctionEndInfo info) { System.out.printf( - "[PLUGIN] onUserFunctionEnd: name=%s, succeeded=%s, error=%s%n", + "[PLUGIN] onUserFunctionEnd: name=%s, outcome=%s, error=%s%n", info.name(), - info.succeeded(), + info.outcome(), info.error() != null ? info.error().getMessage() : null); } } diff --git a/otel-plugin/README.md b/otel-plugin/README.md index dcf5d1b4d..107354f57 100644 --- a/otel-plugin/README.md +++ b/otel-plugin/README.md @@ -211,7 +211,7 @@ Operation and attempt spans link to the Workflow span. `ExecutionOtelPlugin` rev | `durable.operation.type` | Parent operation type | | `durable.operation.name` | Parent operation name | | `durable.attempt.number` | 1-based attempt number | -| `durable.attempt.outcome` | SUCCEEDED or FAILED | +| `durable.attempt.outcome` | SUCCEEDED (span status `OK`), FAILED (`ERROR`), or INCOMPLETE (`UNSET`) | ## Log Correlation (MDC) diff --git a/otel-plugin/src/main/java/software/amazon/lambda/durable/otel/ExecutionOtelPlugin.java b/otel-plugin/src/main/java/software/amazon/lambda/durable/otel/ExecutionOtelPlugin.java index 23ebda029..8170ce008 100644 --- a/otel-plugin/src/main/java/software/amazon/lambda/durable/otel/ExecutionOtelPlugin.java +++ b/otel-plugin/src/main/java/software/amazon/lambda/durable/otel/ExecutionOtelPlugin.java @@ -495,14 +495,19 @@ public void onUserFunctionEnd(UserFunctionEndInfo info) { var span = attemptSpans.remove(key); if (span == null) return; - var outcome = info.succeeded() ? "SUCCEEDED" : "FAILED"; - span.setAttribute(DURABLE_ATTEMPT_OUTCOME, outcome); - - if (!info.succeeded() && info.error() != null) { - span.setStatus(StatusCode.ERROR, info.error().getMessage()); - span.recordException(info.error()); - } else if (info.succeeded()) { - span.setStatus(StatusCode.OK); + span.setAttribute(DURABLE_ATTEMPT_OUTCOME, info.outcome().name()); + + switch (info.outcome()) { + case SUCCEEDED -> span.setStatus(StatusCode.OK); + case FAILED -> { + if (info.error() != null) { + span.setStatus(StatusCode.ERROR, info.error().getMessage()); + span.recordException(info.error()); + } + } + case INCOMPLETE -> { + // An incomplete user function is expected when the durable execution suspends. + } } endSpan(span, info.endTimestamp()); diff --git a/otel-plugin/src/main/java/software/amazon/lambda/durable/otel/InvocationOtelPlugin.java b/otel-plugin/src/main/java/software/amazon/lambda/durable/otel/InvocationOtelPlugin.java index 83d962306..75df1f60b 100644 --- a/otel-plugin/src/main/java/software/amazon/lambda/durable/otel/InvocationOtelPlugin.java +++ b/otel-plugin/src/main/java/software/amazon/lambda/durable/otel/InvocationOtelPlugin.java @@ -523,14 +523,19 @@ public void onUserFunctionEnd(UserFunctionEndInfo info) { var span = attemptSpans.remove(key); if (span == null) return; - var outcome = info.succeeded() ? "SUCCEEDED" : "FAILED"; - span.setAttribute(DURABLE_ATTEMPT_OUTCOME, outcome); - - if (!info.succeeded() && info.error() != null) { - span.setStatus(StatusCode.ERROR, info.error().getMessage()); - span.recordException(info.error()); - } else if (info.succeeded()) { - span.setStatus(StatusCode.OK); + span.setAttribute(DURABLE_ATTEMPT_OUTCOME, info.outcome().name()); + + switch (info.outcome()) { + case SUCCEEDED -> span.setStatus(StatusCode.OK); + case FAILED -> { + if (info.error() != null) { + span.setStatus(StatusCode.ERROR, info.error().getMessage()); + span.recordException(info.error()); + } + } + case INCOMPLETE -> { + // An incomplete user function is expected when the durable execution suspends. + } } if (info.endTimestamp() != null) { diff --git a/otel-plugin/src/test/java/software/amazon/lambda/durable/otel/ExecutionOtelPluginTest.java b/otel-plugin/src/test/java/software/amazon/lambda/durable/otel/ExecutionOtelPluginTest.java index 0c2a1d3f1..887e86666 100644 --- a/otel-plugin/src/test/java/software/amazon/lambda/durable/otel/ExecutionOtelPluginTest.java +++ b/otel-plugin/src/test/java/software/amazon/lambda/durable/otel/ExecutionOtelPluginTest.java @@ -23,6 +23,7 @@ import org.junit.jupiter.api.AfterEach; import org.junit.jupiter.api.BeforeEach; import org.junit.jupiter.api.Test; +import software.amazon.lambda.durable.execution.SuspendExecutionException; import software.amazon.lambda.durable.plugin.*; class ExecutionOtelPluginTest { @@ -459,7 +460,17 @@ void attemptSpan_childOfOperation_linkedToInvocation() { plugin.onUserFunctionStart( new UserFunctionStartInfo("op-1", "compute", "STEP", "Step", null, Instant.now(), false, 1)); plugin.onUserFunctionEnd(new UserFunctionEndInfo( - "op-1", "compute", "STEP", "Step", null, Instant.now(), Instant.now(), false, 1, true, null)); + "op-1", + "compute", + "STEP", + "Step", + null, + Instant.now(), + Instant.now(), + false, + 1, + UserFunctionOutcome.SUCCEEDED, + null)); plugin.onOperationEnd(new OperationEndInfo( "op-1", "compute", @@ -500,7 +511,17 @@ void attemptSpan_carriesOperationSubtype() { plugin.onUserFunctionStart( new UserFunctionStartInfo("op-1", "process-order", "STEP", "Step", null, Instant.now(), false, 1)); plugin.onUserFunctionEnd(new UserFunctionEndInfo( - "op-1", "process-order", "STEP", "Step", null, Instant.now(), Instant.now(), false, 1, true, null)); + "op-1", + "process-order", + "STEP", + "Step", + null, + Instant.now(), + Instant.now(), + false, + 1, + UserFunctionOutcome.SUCCEEDED, + null)); plugin.onInvocationEnd(new InvocationEndInfo("req-1", ARN, true, InvocationStatus.SUCCEEDED, null)); var attemptSpan = spanExporter.getFinishedSpanItems().stream() @@ -574,7 +595,7 @@ void userFunctionFailure_setsErrorOnAttemptSpan() { Instant.now(), false, 1, - false, + UserFunctionOutcome.FAILED, new RuntimeException("step failed"))); plugin.onInvocationEnd(new InvocationEndInfo("req-1", ARN, true, InvocationStatus.FAILED, null)); @@ -591,7 +612,17 @@ void userFunctionSuccess_setsOkOnAttemptSpan() { plugin.onUserFunctionStart( new UserFunctionStartInfo("op-1", "compute", "STEP", "Step", null, Instant.now(), false, 1)); plugin.onUserFunctionEnd(new UserFunctionEndInfo( - "op-1", "compute", "STEP", "Step", null, Instant.now(), Instant.now(), false, 1, true, null)); + "op-1", + "compute", + "STEP", + "Step", + null, + Instant.now(), + Instant.now(), + false, + 1, + UserFunctionOutcome.SUCCEEDED, + null)); plugin.onInvocationEnd(new InvocationEndInfo("req-1", ARN, true, InvocationStatus.SUCCEEDED, null)); var attemptSpan = spanExporter.getFinishedSpanItems().stream() @@ -601,6 +632,31 @@ void userFunctionSuccess_setsOkOnAttemptSpan() { assertEquals(StatusCode.OK, attemptSpan.getStatus().getStatusCode()); } + @Test + void userFunctionIncomplete_leavesAttemptSpanUnset() { + plugin.onInvocationStart(new InvocationInfo("req-1", ARN, true, Instant.now())); + plugin.onUserFunctionStart( + new UserFunctionStartInfo("op-1", "waiting", "STEP", "Step", null, Instant.now(), false, 1)); + plugin.onUserFunctionEnd(new UserFunctionEndInfo( + "op-1", + "waiting", + "STEP", + "Step", + null, + Instant.now(), + Instant.now(), + false, + 1, + UserFunctionOutcome.INCOMPLETE, + new SuspendExecutionException())); + plugin.onInvocationEnd(new InvocationEndInfo("req-1", ARN, true, InvocationStatus.PENDING, null)); + + var attemptSpan = spanByName(spanExporter.getFinishedSpanItems(), "waiting attempt 1"); + assertEquals(StatusCode.UNSET, attemptSpan.getStatus().getStatusCode()); + assertEquals("INCOMPLETE", attemptSpan.getAttributes().get(AttributeKey.stringKey("durable.attempt.outcome"))); + assertTrue(attemptSpan.getEvents().isEmpty()); + } + @Test void operationSuccess_setsOkOnOperationSpan() { plugin.onInvocationStart(new InvocationInfo("req-1", ARN, true, Instant.now())); diff --git a/otel-plugin/src/test/java/software/amazon/lambda/durable/otel/InvocationOtelPluginTest.java b/otel-plugin/src/test/java/software/amazon/lambda/durable/otel/InvocationOtelPluginTest.java index be68c9716..0d14daa15 100644 --- a/otel-plugin/src/test/java/software/amazon/lambda/durable/otel/InvocationOtelPluginTest.java +++ b/otel-plugin/src/test/java/software/amazon/lambda/durable/otel/InvocationOtelPluginTest.java @@ -452,7 +452,17 @@ void attemptSpanName_usesOperationNameWithAttemptNumber() { plugin.onUserFunctionStart( new UserFunctionStartInfo("op-1", "process-order", "STEP", "Step", null, Instant.now(), false, 1)); plugin.onUserFunctionEnd(new UserFunctionEndInfo( - "op-1", "process-order", "STEP", "Step", null, Instant.now(), Instant.now(), false, 1, true, null)); + "op-1", + "process-order", + "STEP", + "Step", + null, + Instant.now(), + Instant.now(), + false, + 1, + UserFunctionOutcome.SUCCEEDED, + null)); plugin.onInvocationEnd(new InvocationEndInfo("req-1", "arn:exec1", true, InvocationStatus.SUCCEEDED, null)); var attemptSpan = spanExporter.getFinishedSpanItems().stream() @@ -502,7 +512,17 @@ void attemptSpan_carriesOperationSubtype() { plugin.onUserFunctionStart( new UserFunctionStartInfo("op-1", "process-order", "STEP", "Step", null, Instant.now(), false, 1)); plugin.onUserFunctionEnd(new UserFunctionEndInfo( - "op-1", "process-order", "STEP", "Step", null, Instant.now(), Instant.now(), false, 1, true, null)); + "op-1", + "process-order", + "STEP", + "Step", + null, + Instant.now(), + Instant.now(), + false, + 1, + UserFunctionOutcome.SUCCEEDED, + null)); plugin.onInvocationEnd(new InvocationEndInfo("req-1", "arn:exec1", true, InvocationStatus.SUCCEEDED, null)); var attemptSpan = spanExporter.getFinishedSpanItems().stream() @@ -609,7 +629,17 @@ void userFunctionStart_and_end_createsAttemptSpan() { new UserFunctionStartInfo("op-1", "compute", "STEP", "Step", null, Instant.now(), false, 1)); plugin.onUserFunctionEnd(new UserFunctionEndInfo( - "op-1", "compute", "STEP", "Step", null, Instant.now(), Instant.now(), false, 1, true, null)); + "op-1", + "compute", + "STEP", + "Step", + null, + Instant.now(), + Instant.now(), + false, + 1, + UserFunctionOutcome.SUCCEEDED, + null)); plugin.onInvocationEnd(new InvocationEndInfo("req-1", "arn:exec1", true, InvocationStatus.SUCCEEDED, null)); @@ -641,7 +671,7 @@ void userFunctionEnd_withFailure_setsErrorOnAttemptSpan() { Instant.now(), false, 1, - false, + UserFunctionOutcome.FAILED, new RuntimeException("step failed"))); plugin.onInvocationEnd(new InvocationEndInfo("req-1", "arn:exec1", true, InvocationStatus.FAILED, null)); @@ -660,7 +690,17 @@ void userFunctionEnd_withSuccess_setsOkOnAttemptSpan() { plugin.onUserFunctionStart( new UserFunctionStartInfo("op-1", "compute", "STEP", "Step", null, Instant.now(), false, 1)); plugin.onUserFunctionEnd(new UserFunctionEndInfo( - "op-1", "compute", "STEP", "Step", null, Instant.now(), Instant.now(), false, 1, true, null)); + "op-1", + "compute", + "STEP", + "Step", + null, + Instant.now(), + Instant.now(), + false, + 1, + UserFunctionOutcome.SUCCEEDED, + null)); plugin.onInvocationEnd(new InvocationEndInfo("req-1", "arn:exec1", true, InvocationStatus.SUCCEEDED, null)); @@ -671,6 +711,36 @@ void userFunctionEnd_withSuccess_setsOkOnAttemptSpan() { assertEquals(StatusCode.OK, attemptSpan.getStatus().getStatusCode()); } + @Test + void userFunctionEnd_withIncomplete_leavesAttemptSpanUnset() { + plugin.onInvocationStart(new InvocationInfo("req-1", "arn:exec1", true, Instant.now())); + + plugin.onUserFunctionStart( + new UserFunctionStartInfo("op-1", "waiting", "STEP", "Step", null, Instant.now(), false, 1)); + plugin.onUserFunctionEnd(new UserFunctionEndInfo( + "op-1", + "waiting", + "STEP", + "Step", + null, + Instant.now(), + Instant.now(), + false, + 1, + UserFunctionOutcome.INCOMPLETE, + new SuspendExecutionException())); + + plugin.onInvocationEnd(new InvocationEndInfo("req-1", "arn:exec1", true, InvocationStatus.PENDING, null)); + + var attemptSpan = spanExporter.getFinishedSpanItems().stream() + .filter(s -> s.getName().equals("waiting attempt 1")) + .findFirst() + .orElseThrow(); + assertEquals(StatusCode.UNSET, attemptSpan.getStatus().getStatusCode()); + assertEquals("INCOMPLETE", attemptSpan.getAttributes().get(AttributeKey.stringKey("durable.attempt.outcome"))); + assertTrue(attemptSpan.getEvents().isEmpty()); + } + @Test void operationEnd_withSuccess_setsOkOnOperationSpan() { plugin.onInvocationStart(new InvocationInfo("req-1", "arn:exec1", true, Instant.now())); @@ -805,7 +875,17 @@ void fullLifecycle_producesCorrectSpanHierarchy() { plugin.onUserFunctionStart( new UserFunctionStartInfo("op-1", "step-a", "STEP", "Step", null, Instant.now(), false, 1)); plugin.onUserFunctionEnd(new UserFunctionEndInfo( - "op-1", "step-a", "STEP", "Step", null, Instant.now(), Instant.now(), false, 1, true, null)); + "op-1", + "step-a", + "STEP", + "Step", + null, + Instant.now(), + Instant.now(), + false, + 1, + UserFunctionOutcome.SUCCEEDED, + null)); plugin.onOperationEnd(new OperationEndInfo( "op-1", "step-a", @@ -826,7 +906,17 @@ void fullLifecycle_producesCorrectSpanHierarchy() { plugin.onUserFunctionStart( new UserFunctionStartInfo("op-2", "step-b", "STEP", "Step", null, Instant.now(), false, 1)); plugin.onUserFunctionEnd(new UserFunctionEndInfo( - "op-2", "step-b", "STEP", "Step", null, Instant.now(), Instant.now(), false, 1, true, null)); + "op-2", + "step-b", + "STEP", + "Step", + null, + Instant.now(), + Instant.now(), + false, + 1, + UserFunctionOutcome.SUCCEEDED, + null)); plugin.onOperationEnd(new OperationEndInfo( "op-2", "step-b", @@ -963,7 +1053,17 @@ void sampling_disabled_producesNoSpans() { sampledPlugin.onUserFunctionStart( new UserFunctionStartInfo("op-1", "step", "STEP", "Step", null, Instant.now(), false, 1)); sampledPlugin.onUserFunctionEnd(new UserFunctionEndInfo( - "op-1", "step", "STEP", "Step", null, Instant.now(), Instant.now(), false, 1, true, null)); + "op-1", + "step", + "STEP", + "Step", + null, + Instant.now(), + Instant.now(), + false, + 1, + UserFunctionOutcome.SUCCEEDED, + null)); sampledPlugin.onOperationEnd(new OperationEndInfo( "op-1", "step", @@ -1031,7 +1131,17 @@ void xrayExtraction_invocationTreeUsesExtractedTraceId_workflowRemainsIndependen xrayPlugin.onUserFunctionStart( new UserFunctionStartInfo("op-1", "step-a", "STEP", "Step", null, Instant.now(), false, 1)); xrayPlugin.onUserFunctionEnd(new UserFunctionEndInfo( - "op-1", "step-a", "STEP", "Step", null, Instant.now(), Instant.now(), false, 1, true, null)); + "op-1", + "step-a", + "STEP", + "Step", + null, + Instant.now(), + Instant.now(), + false, + 1, + UserFunctionOutcome.SUCCEEDED, + null)); xrayPlugin.onOperationEnd(new OperationEndInfo( "op-1", "step-a", @@ -1350,7 +1460,7 @@ void contextOperation_doesNotCreateAttemptSpan() { Instant.now(), false, null, - false, + UserFunctionOutcome.INCOMPLETE, new SuspendExecutionException())); plugin.onInvocationEnd(new InvocationEndInfo("req-1", "arn:exec1", true, InvocationStatus.PENDING, null)); @@ -1456,7 +1566,17 @@ void multiInvocation_stepWaitStep_producesCorrectSpans() { plugin.onUserFunctionStart( new UserFunctionStartInfo("op-1", "step-A", "STEP", "Step", null, Instant.now(), false, 1)); plugin.onUserFunctionEnd(new UserFunctionEndInfo( - "op-1", "step-A", "STEP", "Step", null, Instant.now(), Instant.now(), false, 1, true, null)); + "op-1", + "step-A", + "STEP", + "Step", + null, + Instant.now(), + Instant.now(), + false, + 1, + UserFunctionOutcome.SUCCEEDED, + null)); plugin.onOperationEnd(new OperationEndInfo( "op-1", "step-A", @@ -1510,7 +1630,17 @@ void multiInvocation_stepWaitStep_producesCorrectSpans() { plugin.onUserFunctionStart( new UserFunctionStartInfo("op-3", "step-B", "STEP", "Step", null, Instant.now(), false, 1)); plugin.onUserFunctionEnd(new UserFunctionEndInfo( - "op-3", "step-B", "STEP", "Step", null, Instant.now(), Instant.now(), false, 1, true, null)); + "op-3", + "step-B", + "STEP", + "Step", + null, + Instant.now(), + Instant.now(), + false, + 1, + UserFunctionOutcome.SUCCEEDED, + null)); plugin.onOperationEnd(new OperationEndInfo( "op-3", "step-B", @@ -1577,7 +1707,7 @@ void crossInvocation_stepRetry_attemptsParentedToRespectiveInvocations() { Instant.now(), false, 1, - false, + UserFunctionOutcome.FAILED, new RuntimeException("payment failed"))); plugin.onInvocationEnd(new InvocationEndInfo("req-1", arn, true, InvocationStatus.PENDING, null)); @@ -1611,7 +1741,17 @@ void crossInvocation_stepRetry_attemptsParentedToRespectiveInvocations() { plugin.onUserFunctionStart( new UserFunctionStartInfo("op-1", "process-payment", "STEP", "Step", null, Instant.now(), false, 2)); plugin.onUserFunctionEnd(new UserFunctionEndInfo( - "op-1", "process-payment", "STEP", "Step", null, Instant.now(), Instant.now(), false, 2, true, null)); + "op-1", + "process-payment", + "STEP", + "Step", + null, + Instant.now(), + Instant.now(), + false, + 2, + UserFunctionOutcome.SUCCEEDED, + null)); plugin.onOperationEnd(new OperationEndInfo( "op-1", "process-payment", @@ -1713,7 +1853,17 @@ void operationAndAttemptSpans_linkToWorkflowSpan() { plugin.onUserFunctionStart( new UserFunctionStartInfo("op-1", "step-a", "STEP", "Step", null, Instant.now(), false, 1)); plugin.onUserFunctionEnd(new UserFunctionEndInfo( - "op-1", "step-a", "STEP", "Step", null, Instant.now(), Instant.now(), false, 1, true, null)); + "op-1", + "step-a", + "STEP", + "Step", + null, + Instant.now(), + Instant.now(), + false, + 1, + UserFunctionOutcome.SUCCEEDED, + null)); plugin.onOperationEnd(new OperationEndInfo( "op-1", "step-a", diff --git a/otel-plugin/src/test/java/software/amazon/lambda/durable/otel/MdcSpanEnricherTest.java b/otel-plugin/src/test/java/software/amazon/lambda/durable/otel/MdcSpanEnricherTest.java index 2318e4075..9c4399817 100644 --- a/otel-plugin/src/test/java/software/amazon/lambda/durable/otel/MdcSpanEnricherTest.java +++ b/otel-plugin/src/test/java/software/amazon/lambda/durable/otel/MdcSpanEnricherTest.java @@ -74,7 +74,17 @@ void plugin_withMdcEnabled_setsFieldsInMdc() { assertNotNull(MDC.get(MdcSpanEnricher.MDC_TRACE_SAMPLED)); plugin.onUserFunctionEnd(new UserFunctionEndInfo( - "op-1", "step", "STEP", "Step", null, Instant.now(), Instant.now(), false, 1, true, null)); + "op-1", + "step", + "STEP", + "Step", + null, + Instant.now(), + Instant.now(), + false, + 1, + UserFunctionOutcome.SUCCEEDED, + null)); // After onUserFunctionEnd: span_id is cleared, but trace_id remains for handler-level logs between steps assertNotNull(MDC.get(MdcSpanEnricher.MDC_TRACE_ID), "trace_id should persist between steps"); diff --git a/sdk-integration-tests/src/test/java/software/amazon/lambda/durable/PluginIntegrationTest.java b/sdk-integration-tests/src/test/java/software/amazon/lambda/durable/PluginIntegrationTest.java index 7da27558e..dfe99b9d7 100644 --- a/sdk-integration-tests/src/test/java/software/amazon/lambda/durable/PluginIntegrationTest.java +++ b/sdk-integration-tests/src/test/java/software/amazon/lambda/durable/PluginIntegrationTest.java @@ -10,6 +10,7 @@ import java.util.Collections; import java.util.List; import java.util.Map; +import java.util.concurrent.CompletionException; import java.util.concurrent.ConcurrentHashMap; import java.util.concurrent.atomic.AtomicInteger; import java.util.concurrent.atomic.AtomicReference; @@ -648,17 +649,19 @@ void plugin_receivesUserFunctionStartAndEnd_forStep() { plugin.userFunctionStarts.stream().anyMatch(info -> "compute".equals(info.name())), "Should have onUserFunctionStart for 'compute'"); assertTrue( - plugin.userFunctionEnds.stream().anyMatch(info -> "compute".equals(info.name()) && info.succeeded()), + plugin.userFunctionEnds.stream() + .anyMatch(info -> + "compute".equals(info.name()) && info.outcome() == UserFunctionOutcome.SUCCEEDED), "Should have successful onUserFunctionEnd for 'compute'"); } @Test - void plugin_userFunctionEnd_succeeded_false_whenStepFails() { + void plugin_userFunctionEnd_reportsFailed_whenStepFails() { var plugin = new RecordingPlugin(); var config = DurableConfig.builder().withPlugins(plugin).build(); // When a step's user function throws, the exception propagates through the user-function hook - // boundary, so onUserFunctionEnd reports succeeded=false with the error. Retry/checkpoint + // boundary, so onUserFunctionEnd reports FAILED with the error. Retry/checkpoint // handling happens outside that boundary. var runner = LocalDurableTestRunner.create( String.class, @@ -688,7 +691,7 @@ void plugin_userFunctionEnd_succeeded_false_whenStepFails() { .filter(info -> "fail-step".equals(info.name())) .findFirst() .orElseThrow(() -> new AssertionError("Should have user function end for 'fail-step'")); - assertFalse(failStepEnd.succeeded(), "Failed step should report succeeded=false"); + assertEquals(UserFunctionOutcome.FAILED, failStepEnd.outcome()); assertNotNull(failStepEnd.error()); assertTrue(failStepEnd.error().getMessage().contains("step failed")); } @@ -860,7 +863,7 @@ void plugin_reportsFailedThenSucceededAttempts_forRetriedStep() { .filter(info -> info.attempt() == 1) .findFirst() .orElseThrow(); - assertFalse(firstAttempt.succeeded()); + assertEquals(UserFunctionOutcome.FAILED, firstAttempt.outcome()); assertNotNull(firstAttempt.error()); // Retry attempt succeeded. @@ -868,7 +871,7 @@ void plugin_reportsFailedThenSucceededAttempts_forRetriedStep() { .filter(info -> info.attempt() == 2) .findFirst() .orElseThrow(); - assertTrue(secondAttempt.succeeded()); + assertEquals(UserFunctionOutcome.SUCCEEDED, secondAttempt.outcome()); assertNull(secondAttempt.error()); // The operation end reports the terminal attempt number = total attempts that ran. The step @@ -881,12 +884,12 @@ void plugin_reportsFailedThenSucceededAttempts_forRetriedStep() { } @Test - void plugin_userFunctionEnd_reportsSuspension_asNotSucceeded() { + void plugin_userFunctionEnd_reportsSuspension_asIncomplete() { var plugin = new RecordingPlugin(); var config = DurableConfig.builder().withPlugins(plugin).build(); // A child context whose body suspends (on a wait) throws SuspendExecutionException through the - // user-function boundary, so onUserFunctionEnd fires with succeeded=false and the suspend exception. + // user-function boundary, so onUserFunctionEnd fires with INCOMPLETE and the suspend exception. var runner = LocalDurableTestRunner.create( String.class, (input, context) -> context.runInChildContext("child", TypeToken.get(String.class), child -> { @@ -902,12 +905,36 @@ void plugin_userFunctionEnd_reportsSuspension_asNotSucceeded() { .filter(info -> "child".equals(info.name())) .findFirst() .orElseThrow(() -> new AssertionError("onUserFunctionEnd should fire for the suspended child context")); - assertFalse(childEnd.succeeded(), "A suspended user function must not report succeeded=true"); + assertEquals(UserFunctionOutcome.INCOMPLETE, childEnd.outcome()); assertTrue( childEnd.error() instanceof SuspendExecutionException, "Suspension should surface the SuspendExecutionException, not a user error"); } + @Test + void plugin_userFunctionEnd_unwrapsCompletionExceptionForSuspension() { + var plugin = new RecordingPlugin(); + var config = DurableConfig.builder().withPlugins(plugin).build(); + var suspension = new SuspendExecutionException(); + + var runner = LocalDurableTestRunner.create( + String.class, + (input, context) -> context.step("wrapped-suspension", String.class, step -> { + throw new CompletionException(suspension); + }), + config); + + var result = runner.run("input"); + assertEquals(ExecutionStatus.PENDING, result.getStatus()); + + var stepEnd = plugin.userFunctionEnds.stream() + .filter(info -> "wrapped-suspension".equals(info.name())) + .findFirst() + .orElseThrow(() -> new AssertionError("onUserFunctionEnd should fire for the suspended step")); + assertEquals(UserFunctionOutcome.INCOMPLETE, stepEnd.outcome()); + assertSame(suspension, stepEnd.error(), "The event should expose the unwrapped suspension"); + } + @Test void plugin_parallelBranches_emitUserFunctionHooks_butConsumerDoesNot() { var plugin = new RecordingPlugin(); diff --git a/sdk/src/main/java/software/amazon/lambda/durable/operation/BaseDurableOperation.java b/sdk/src/main/java/software/amazon/lambda/durable/operation/BaseDurableOperation.java index f0391567a..35a71f0da 100644 --- a/sdk/src/main/java/software/amazon/lambda/durable/operation/BaseDurableOperation.java +++ b/sdk/src/main/java/software/amazon/lambda/durable/operation/BaseDurableOperation.java @@ -29,6 +29,7 @@ import software.amazon.lambda.durable.model.OperationSubType; import software.amazon.lambda.durable.plugin.PluginInfoConverter; import software.amazon.lambda.durable.plugin.PluginRunner; +import software.amazon.lambda.durable.plugin.UserFunctionOutcome; import software.amazon.lambda.durable.util.ExceptionHelper; /** @@ -354,10 +355,18 @@ protected T runUserFunction(Integer attempt, Supplier userFunction) { pluginRunner.onUserFunctionStart(startInfo); try { T result = userFunction.get(); - pluginRunner.onUserFunctionEnd(PluginInfoConverter.toUserFunctionEndInfo(startInfo, true, null)); + pluginRunner.onUserFunctionEnd( + PluginInfoConverter.toUserFunctionEndInfo(startInfo, UserFunctionOutcome.SUCCEEDED, null)); return result; } catch (Throwable e) { - pluginRunner.onUserFunctionEnd(PluginInfoConverter.toUserFunctionEndInfo(startInfo, false, e)); + var error = ExceptionHelper.unwrapCompletableFuture(e); + if (error == null) { + error = e; + } + var outcome = error instanceof SuspendExecutionException + ? UserFunctionOutcome.INCOMPLETE + : UserFunctionOutcome.FAILED; + pluginRunner.onUserFunctionEnd(PluginInfoConverter.toUserFunctionEndInfo(startInfo, outcome, error)); ExceptionHelper.sneakyThrow(e); return null; // unreachable — sneakyThrow always throws } diff --git a/sdk/src/main/java/software/amazon/lambda/durable/plugin/DurableExecutionPlugin.java b/sdk/src/main/java/software/amazon/lambda/durable/plugin/DurableExecutionPlugin.java index 92d2b1c6b..e2f8a46df 100644 --- a/sdk/src/main/java/software/amazon/lambda/durable/plugin/DurableExecutionPlugin.java +++ b/sdk/src/main/java/software/amazon/lambda/durable/plugin/DurableExecutionPlugin.java @@ -79,11 +79,10 @@ default void onUserFunctionStart(UserFunctionStartInfo info) {} * *

This hook fires on the same thread as user code, so plugins can close OTel scopes here. * - *

It fires for every terminal outcome of the user function: normal return, a thrown failure, and suspension. - * {@link UserFunctionEndInfo#succeeded()} is {@code true} only on normal return; it is {@code false} for both a - * genuine failure and a suspension. On suspension the {@link UserFunctionEndInfo#error()} is the SDK's internal - * {@code SuspendExecutionException}, which is not a user error — plugins that count failures should treat a - * suspension as a neutral outcome (for example by checking the error type) rather than a failure. + *

It fires for every outcome of the user function: normal return, a thrown failure, and suspension. Check + * {@link UserFunctionEndInfo#outcome()} to distinguish them. A suspended function reports + * {@link UserFunctionOutcome#INCOMPLETE}; its {@link UserFunctionEndInfo#error()} is the SDK's internal + * {@code SuspendExecutionException}, not a user error. */ default void onUserFunctionEnd(UserFunctionEndInfo info) {} } diff --git a/sdk/src/main/java/software/amazon/lambda/durable/plugin/PluginInfoConverter.java b/sdk/src/main/java/software/amazon/lambda/durable/plugin/PluginInfoConverter.java index 59b9e4a3a..131420f39 100644 --- a/sdk/src/main/java/software/amazon/lambda/durable/plugin/PluginInfoConverter.java +++ b/sdk/src/main/java/software/amazon/lambda/durable/plugin/PluginInfoConverter.java @@ -125,12 +125,12 @@ public static UserFunctionStartInfo toUserFunctionStartInfo( * Creates a {@link UserFunctionEndInfo} from a start info and outcome. * * @param startInfo the start info from when the function began - * @param succeeded true if the function completed without error - * @param error the error if the function failed (may be null) + * @param outcome the user function outcome + * @param error the error if the function failed or exited incompletely (may be null) * @return a UserFunctionEndInfo record */ public static UserFunctionEndInfo toUserFunctionEndInfo( - UserFunctionStartInfo startInfo, boolean succeeded, Throwable error) { + UserFunctionStartInfo startInfo, UserFunctionOutcome outcome, Throwable error) { return new UserFunctionEndInfo( startInfo.id(), startInfo.name(), @@ -141,7 +141,7 @@ public static UserFunctionEndInfo toUserFunctionEndInfo( Instant.now(), startInfo.isReplayingChildren(), startInfo.attempt(), - succeeded, + outcome, error); } diff --git a/sdk/src/main/java/software/amazon/lambda/durable/plugin/UserFunctionEndInfo.java b/sdk/src/main/java/software/amazon/lambda/durable/plugin/UserFunctionEndInfo.java index 77cab7cea..cc8b41ac7 100644 --- a/sdk/src/main/java/software/amazon/lambda/durable/plugin/UserFunctionEndInfo.java +++ b/sdk/src/main/java/software/amazon/lambda/durable/plugin/UserFunctionEndInfo.java @@ -19,8 +19,8 @@ * @param endTimestamp when the user function ended * @param isReplayingChildren true if child operations within this context are being replayed from checkpoints * @param attempt 1-based attempt number for steps/waitForCondition, null for context operations - * @param succeeded true if the user function completed without error - * @param error non-null if the user function failed; this component is experimental + * @param outcome the user function outcome + * @param error non-null if the user function failed or exited incompletely; this component is experimental */ public record UserFunctionEndInfo( String id, @@ -32,5 +32,5 @@ public record UserFunctionEndInfo( Instant endTimestamp, boolean isReplayingChildren, Integer attempt, - boolean succeeded, + UserFunctionOutcome outcome, @Experimental Throwable error) {} diff --git a/sdk/src/main/java/software/amazon/lambda/durable/plugin/UserFunctionOutcome.java b/sdk/src/main/java/software/amazon/lambda/durable/plugin/UserFunctionOutcome.java new file mode 100644 index 000000000..0b8d62bb2 --- /dev/null +++ b/sdk/src/main/java/software/amazon/lambda/durable/plugin/UserFunctionOutcome.java @@ -0,0 +1,13 @@ +// Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved. +// SPDX-License-Identifier: Apache-2.0 +package software.amazon.lambda.durable.plugin; + +/** Outcome of a user-provided function when its execution ends. */ +public enum UserFunctionOutcome { + /** The user function returned normally. */ + SUCCEEDED, + /** The user function threw an error. */ + FAILED, + /** The user function exited before completing, for example when the durable execution suspended. */ + INCOMPLETE +} diff --git a/sdk/src/test/java/software/amazon/lambda/durable/plugin/PluginInfoConverterTest.java b/sdk/src/test/java/software/amazon/lambda/durable/plugin/PluginInfoConverterTest.java index b4686a16b..152ef0a4d 100644 --- a/sdk/src/test/java/software/amazon/lambda/durable/plugin/PluginInfoConverterTest.java +++ b/sdk/src/test/java/software/amazon/lambda/durable/plugin/PluginInfoConverterTest.java @@ -13,6 +13,7 @@ import software.amazon.awssdk.services.lambda.model.OperationStatus; import software.amazon.awssdk.services.lambda.model.OperationType; import software.amazon.awssdk.services.lambda.model.StepDetails; +import software.amazon.lambda.durable.execution.SuspendExecutionException; import software.amazon.lambda.durable.model.OperationIdentifier; import software.amazon.lambda.durable.model.OperationSubType; @@ -259,7 +260,7 @@ void toUserFunctionStartInfo_contextOperation() { void toUserFunctionEndInfo_succeeded() { var startInfo = PluginInfoConverter.toUserFunctionStartInfo(STEP_IDENTIFIER, PARENT_ID, false, 1); - var endInfo = PluginInfoConverter.toUserFunctionEndInfo(startInfo, true, null); + var endInfo = PluginInfoConverter.toUserFunctionEndInfo(startInfo, UserFunctionOutcome.SUCCEEDED, null); assertEquals(OPERATION_ID, endInfo.id()); assertEquals(OPERATION_NAME, endInfo.name()); @@ -267,7 +268,7 @@ void toUserFunctionEndInfo_succeeded() { assertNotNull(endInfo.endTimestamp()); assertFalse(endInfo.isReplayingChildren()); assertEquals(1, endInfo.attempt()); - assertTrue(endInfo.succeeded()); + assertEquals(UserFunctionOutcome.SUCCEEDED, endInfo.outcome()); assertNull(endInfo.error()); } @@ -276,10 +277,21 @@ void toUserFunctionEndInfo_failed() { var error = new RuntimeException("step failed"); var startInfo = PluginInfoConverter.toUserFunctionStartInfo(STEP_IDENTIFIER, null, false, 2); - var endInfo = PluginInfoConverter.toUserFunctionEndInfo(startInfo, false, error); + var endInfo = PluginInfoConverter.toUserFunctionEndInfo(startInfo, UserFunctionOutcome.FAILED, error); - assertFalse(endInfo.succeeded()); + assertEquals(UserFunctionOutcome.FAILED, endInfo.outcome()); assertEquals(error, endInfo.error()); assertEquals(2, endInfo.attempt()); } + + @Test + void toUserFunctionEndInfo_incomplete() { + var error = new SuspendExecutionException(); + var startInfo = PluginInfoConverter.toUserFunctionStartInfo(MAP_IDENTIFIER, null, false, null); + + var endInfo = PluginInfoConverter.toUserFunctionEndInfo(startInfo, UserFunctionOutcome.INCOMPLETE, error); + + assertEquals(UserFunctionOutcome.INCOMPLETE, endInfo.outcome()); + assertSame(error, endInfo.error()); + } } diff --git a/sdk/src/test/java/software/amazon/lambda/durable/plugin/PluginRunnerTest.java b/sdk/src/test/java/software/amazon/lambda/durable/plugin/PluginRunnerTest.java index bd33b002f..dc21c9e8b 100644 --- a/sdk/src/test/java/software/amazon/lambda/durable/plugin/PluginRunnerTest.java +++ b/sdk/src/test/java/software/amazon/lambda/durable/plugin/PluginRunnerTest.java @@ -243,7 +243,17 @@ private static UserFunctionStartInfo attemptInfo() { private static UserFunctionEndInfo attemptEndInfo() { return new UserFunctionEndInfo( - "op-1", "test-step", "STEP", null, null, Instant.now(), Instant.now(), false, 1, true, null); + "op-1", + "test-step", + "STEP", + null, + null, + Instant.now(), + Instant.now(), + false, + 1, + UserFunctionOutcome.SUCCEEDED, + null); } // ─── Test plugin implementations ─────────────────────────────────────