Skip to content

Commit 166799a

Browse files
dscpinheirodscpinheiro
authored
Durable execution fixes (#2585)
* fix(DurableExecution): treat stale checkpoint token as transient The service rejects a stale token with "Invalid checkpoint token", but IsTerminalCheckpointError matched "Invalid Checkpoint Token" with an ordinal comparison. A stale token was classified as terminal and failed the whole execution instead of letting Lambda retry the invocation. Match case-insensitively and test with the service's real message. Fixes #2571 * fix(DurableExecution): return Pending when checkpoint token is missing A checkpoint response without a CheckpointToken means the service will accept no more checkpoints from this invocation. Previously the next flush sent an empty token, got a 400, and failed the execution. CheckpointBatcher now stops flushing, leaves pending awaiters suspended, and terminates with CheckpointTokenMissing, so the invocation returns Pending and the execution continues in the next invocation. Add TestRunnerOptions.OmitCheckpointToken so workflows can be tested against this path. Fixes #2572 * feat(DurableExecution): add SubType to StepConfig and CallbackConfig Steps and callbacks always recorded "Step"/"Callback", so operations inside a composed child context could only be told apart by name. StepConfig.SubType and CallbackConfig.SubType override the default on every checkpoint; WaitForCallbackConfig applies it to its inner callback. Validate all user-supplied subtypes, including ChildContextConfig.SubType, against the service constraint (1-32 chars from [a-zA-Z0-9-_]) and throw ArgumentException before checkpointing, instead of failing the whole checkpoint request with a 400. Null or empty selects the default. Fixes #2583 * refactor: Update existing docs
1 parent e013684 commit 166799a

28 files changed

Lines changed: 325 additions & 32 deletions

‎.autover/changes/durable-missing-checkpoint-token.json‎

Lines changed: 18 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,18 @@
1+
{
2+
"Projects": [
3+
{
4+
"Name": "Amazon.Lambda.DurableExecution",
5+
"Type": "Patch",
6+
"ChangelogMessages": [
7+
"Return Pending when a checkpoint response has no checkpoint token, instead of failing the execution."
8+
]
9+
},
10+
{
11+
"Name": "Amazon.Lambda.DurableExecution.Testing",
12+
"Type": "Minor",
13+
"ChangelogMessages": [
14+
"Add TestRunnerOptions.OmitCheckpointToken to simulate a checkpoint response without a token."
15+
]
16+
}
17+
]
18+
}

‎.autover/changes/durable-stale-checkpoint-token.json‎

Lines changed: 11 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,11 @@
1+
{
2+
"Projects": [
3+
{
4+
"Name": "Amazon.Lambda.DurableExecution",
5+
"Type": "Patch",
6+
"ChangelogMessages": [
7+
"Treat the service's 'Invalid checkpoint token' error as transient (case-insensitive) so a stale token retries the invocation instead of failing the execution."
8+
]
9+
}
10+
]
11+
}

‎.autover/changes/durable-step-callback-subtype.json‎

Lines changed: 11 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,11 @@
1+
{
2+
"Projects": [
3+
{
4+
"Name": "Amazon.Lambda.DurableExecution",
5+
"Type": "Minor",
6+
"ChangelogMessages": [
7+
"Add SubType to StepConfig and CallbackConfig, and validate SubType values (1 to 32 characters from [a-zA-Z0-9-_]) including ChildContextConfig.SubType."
8+
]
9+
}
10+
]
11+
}

‎Libraries/src/Amazon.Lambda.DurableExecution.Testing/DurableTestRunner.cs‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -59,7 +59,7 @@ internal DurableTestRunner(
5959
_lambdaContext = CreateLambdaContext();
6060
_store = new InMemoryOperationStore();
6161
_processor = new CheckpointProcessor(_store, _options.SkipTime);
62-
_serviceClient = new InMemoryDurableServiceClient(_store, _processor);
62+
_serviceClient = new InMemoryDurableServiceClient(_store, _processor, _options.OmitCheckpointToken);
6363
_registry = registry ?? new FunctionRegistry(_options);
6464
}
6565

‎Libraries/src/Amazon.Lambda.DurableExecution.Testing/FunctionRegistry.cs‎

Lines changed: 1 addition & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -120,6 +120,7 @@ private static string ExtractName(string functionNameOrArn)
120120
var nestedOptions = _options with
121121
{
122122
DurableExecutionArn = _options.DurableExecutionArn + ":nested:" + Guid.NewGuid().ToString("N"),
123+
OmitCheckpointToken = null,
123124
};
124125
var nested = new DurableTestRunner<TPayload, TResult>(handler, nestedOptions, registry: this);
125126

‎Libraries/src/Amazon.Lambda.DurableExecution.Testing/InMemoryDurableServiceClient.cs‎

Lines changed: 7 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -15,11 +15,14 @@ internal sealed class InMemoryDurableServiceClient : IDurableServiceClient
1515
{
1616
private readonly InMemoryOperationStore _store;
1717
private readonly CheckpointProcessor _processor;
18+
private readonly Func<int, bool>? _omitCheckpointToken;
19+
private int _checkpointCalls;
1820

19-
public InMemoryDurableServiceClient(InMemoryOperationStore store, CheckpointProcessor processor)
21+
public InMemoryDurableServiceClient(InMemoryOperationStore store, CheckpointProcessor processor, Func<int, bool>? omitCheckpointToken = null)
2022
{
2123
_store = store;
2224
_processor = processor;
25+
_omitCheckpointToken = omitCheckpointToken;
2326
}
2427

2528
public Task<string?> CheckpointAsync(
@@ -37,6 +40,9 @@ public InMemoryDurableServiceClient(InMemoryOperationStore store, CheckpointProc
3740
if (onNewOperations is not null && newOps.Count > 0)
3841
onNewOperations(newOps);
3942

43+
if (_omitCheckpointToken?.Invoke(Interlocked.Increment(ref _checkpointCalls)) == true)
44+
return Task.FromResult<string?>(null);
45+
4046
return Task.FromResult<string?>(newToken);
4147
}
4248

‎Libraries/src/Amazon.Lambda.DurableExecution.Testing/README.md‎

Lines changed: 1 addition & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -178,6 +178,7 @@ Invoking an unregistered function throws `UnregisteredSiblingFunctionException`.
178178
| `Serializer` | `DefaultLambdaJsonSerializer` | `ILambdaSerializer` for payloads and results. |
179179
| `LoggerFactory` | none | Optional logging during execution. |
180180
| `DurableExecutionArn` | synthetic test ARN | Override for tests that assert on the ARN. |
181+
| `OmitCheckpointToken` | none | Predicate over the 1-based checkpoint call number; when true, the response omits the token so the invocation suspends (`Pending`) and resumes in the next one. |
181182

182183
## Testing Against a Deployed Function
183184

‎Libraries/src/Amazon.Lambda.DurableExecution.Testing/TestRunnerOptions.cs‎

Lines changed: 6 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -62,4 +62,10 @@ public int MaxInvocations
6262
/// assert on ARN values. Default: a synthetic test ARN.
6363
/// </summary>
6464
public string DurableExecutionArn { get; init; } = "arn:aws:lambda:us-east-1:123456789012:execution:test-fn:test-execution";
65+
66+
/// <summary>
67+
/// Given the 1-based checkpoint call number, return true to accept that checkpoint but
68+
/// omit its token from the response, which suspends the invocation. Default: null.
69+
/// </summary>
70+
public Func<int, bool>? OmitCheckpointToken { get; init; }
6571
}

‎Libraries/src/Amazon.Lambda.DurableExecution/CallbackConfig.cs‎

Lines changed: 7 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -71,6 +71,13 @@ public TimeSpan HeartbeatTimeout
7171
/// </summary>
7272
public ILambdaSerializer? Serializer { get; set; }
7373

74+
/// <summary>
75+
/// Optional operation subtype label for observability, recorded on the callback's
76+
/// checkpoint. Null or empty uses the default (<c>"Callback"</c>). Must be 1 to 32
77+
/// characters from <c>[a-zA-Z0-9-_]</c>.
78+
/// </summary>
79+
public string? SubType { get; set; }
80+
7481
private static void ValidateTimeout(TimeSpan value, string paramName)
7582
{
7683
// Allow Zero (means "not set"); reject negative; reject sub-second

‎Libraries/src/Amazon.Lambda.DurableExecution/ChildContextConfig.cs‎

Lines changed: 2 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -18,7 +18,8 @@ public sealed class ChildContextConfig
1818
{
1919
/// <summary>
2020
/// Operation sub-type label for observability (e.g. <c>"WaitForCallback"</c>).
21-
/// Surfaces on the wire <c>OperationUpdate.SubType</c> field.
21+
/// Surfaces on the wire <c>OperationUpdate.SubType</c> field. Null or empty sends
22+
/// no subtype. Must be 1 to 32 characters from <c>[a-zA-Z0-9-_]</c>.
2223
/// </summary>
2324
public string? SubType { get; set; }
2425

0 commit comments

Comments
 (0)