aws / aws/aws-durable-execution-sdk-java
[Bug]: DurableFuture.anyOf blocks without deregistering the active thread, so the execution never suspends
- 主要言語
- Java
- スター
- 28
- フォーク
- 11
- 平均マージ
- 1日 8時間
- マージ済み PR(30日)
- 47
説明
## Expected Behavior
`DurableFuture.anyOf(...)` should permit suspension while waiting, the same as `get()` and `allOf(...)` do. A handler waiting via `anyOf` on operations that suspend, such as `waitForCallback`, should end the current invocation and resume when one of them completes.
## Actual Behavior
`anyOf` bypasses the thread bookkeeping that makes suspension possible. It reads each operation's raw completion future and joins it directly:
```java
static Object anyOf(DurableFuture... futures) {
return CompletableFuture.anyOf(Arrays.stream(futures)
.map(f -> ((BaseDurableOperation) f).getCompletionFuture())
.toArray(CompletableFuture[]::new))
.thenApply(o -> (DurableFuture) o)
.join()
.get();
}
```
Because `join()` is called on the raw future rather than going through `BaseDurableOperation.waitForOperationCompletion()`, `deregisterActiveThread(...)` is never called. Compare `allOf` directly above it, which maps over `DurableFuture::get` and therefore deregisters correctly.
The consequence is stronger than the calling thread simply remaining registered. `ExecutionManager.shouldSuspendExecution()` is the only place the decision to suspend is made, and it is only ever called from inside `deregisterActiveThread(...)`. With no deregistration, that check is never reached at all, so the execution cannot suspend regardless of state. The trailing `.get()` in `anyOf` would take the correct path, but `.join()` has already blocked before control could reach it.
The practical effect is that an execution waiting via `anyOf` stays alive and billed until the function timeout, replays, and blocks again, instead of suspending at zero compute cost. This contradicts the documented behaviour that [waits suspend execution without incurring compute charges](https://docs.aws.amazon.com/lambda/latest/dg/durable-functions.html). For a workload running at high concurrency it also holds that concurrency for the full duration.
Relevant code:
- [`DurableFuture.anyOf`](https://github.com/aws/aws-durable-execution-sdk-java/blob/main/sdk/src/main/java/software/amazon/lambda/durable/DurableFuture.java) — `anyOf` at lines 64-71, `allOf` at lines 40 and 54
- `BaseDurableOperation.waitForOperationCompletion()` at line 225, whose javadoc states the required behaviour: "Deregisters the current thread to allow Lambda suspension if the operation is still in progress, then re-registers when the operation completes." Re-registration is chained at line 243 and `deregisterActiveThread` is called at line 246
- `ExecutionManager.shouldSuspendExecution()` at lines 361-362, called only from `deregisterActiveThread` at line 329 and a sibling at line 353
## Steps to Reproduce
Three tests that are identical except for how they wait on the same two unresolved `waitForCallbackAsync` futures. The two controls suspend and report `ExecutionStatus.PENDING`; the `anyOf` case never returns.
```java
private static WaitForCallbackConfig longTimeout() {
return WaitForCallbackConfig.builder()
.callbackConfig(CallbackConfig.builder().timeout(Duration.ofMinutes(30)).build())
.build();
}
// CONTROL: suspends, terminal status PENDING
DurableFuture f1 = context.waitForCallbackAsync("cb1", String.class, (id, ctx) -> {}, longTimeout());
return f1.get();
// CONTROL: suspends, terminal status PENDING
DurableFuture f1 = context.waitForCallbackAsync("cb1", String.class, (id, ctx) -> {}, longTimeout());
DurableFuture f2 = context.waitForCallbackAsync("cb2", String.class, (id, ctx) -> {}, longTimeout());
return String.join(",", DurableFuture.allOf(f1, f2));
// REPRO: never suspends, blocks indefinitely
DurableFuture f1 = context.waitForCallbackAsync("cb1", String.class, (id, ctx) -> {}, longTimeout());
DurableFuture f2 = context.waitForCallbackAsync("cb2", String.class, (id, ctx) -> {}, longTimeout());
return String.valueOf(DurableFuture.anyOf(f1, f2));
```
Run each with `LocalDurableTestRunner` on a bounded deadline and assert the terminal status. Do not resolve the callbacks.
Result:
```
singleFuture_get_suspends PASSED ExecutionStatus.PENDING
allOf_suspends PASSED ExecutionStatus.PENDING
anyOf_doesNotSuspend FAILED blocked past a 30s deadline, never suspended
```
The same shape reproduces on a deployed function, where Duration and Billed Duration equal the function timeout and concurrency is held throughout, rather than the invocation suspending.
## SDK Version
Reproduced on `main` at 2.2.1-SNAPSHOT. Also present in 2.2.0 and 1.2.1; `DurableFuture.java` is unchanged across them.
## Java Version
21 (also reported on 25)
## Is this a regression?
No
## Additional Context
`BaseDurableOperation.waitForOperationCompletion()` documents the behaviour `anyOf` needs: deregister the calling thread before joining the composed future, and re-register on completion.
Two notes that may be useful:
There appears to be no test covering `anyOf` in `sdk/src/test/java/software/amazon/lambda/durable/DurableFutureTest.java`, which may be why this was not caught. Adding coverage that asserts suspension, not just the returned value, would catch it.
`ParallelConfig.completionConfig(CompletionConfig.firstSuccessful())` is documented for a "first branch wins" pattern, but it is not a substitute where branches are registered incrementally and the caller needs to react to each completion in turn. There is also no workaround available from user code, since the whole public surface of `DurableFuture` is `get()`, the two `allOf` overloads and `anyOf`, and `registerActiveThread` / `deregisterActiveThread` are `protected` on `BaseDurableOperation`. `getCompletionFuture()` is public, but joining it is the bug itself.
Introduced with the feature in #91.
## During the anyOf case, the SDK logs this repeatedly at roughly 1.5 second intervals
```
Calling durable checkpoint API with 0 updates: []
Processing 0 operations. (2 pending pollers)
0 operations processed and 0 pollers completed
```
Two pollers outstanding, nothing to checkpoint, and no suspension.
コントリビューションガイド
調査の方向性
sdk/src/main/java/software/amazon/lambda/durable/DurableFuture.java の DurableFuture.anyOf と allOf から始め、次に BaseDurableOperation.waitForOperationCompletion() と ExecutionManager.shouldSuspendExecution() を読みます。未解決の waitForCallbackAsync futures と LocalDurableTestRunner を使用して、sdk/src/test/java/software/amazon/lambda/durable/DurableFutureTest.java にカバレッジを追加します。anyOf がブロックするのではなくサスペンドし、ExecutionStatus.PENDING に到達すれば完了です。
索引モデルが issue の本文から書いたものです。
評価
- 技術スタック
- aws, java
- 領域
- backend
- issue の種類
- バグ
- 難易度
- 3/5
- 見積もり時間
- 1〜2日
- 活発さ
- 活発
- 明瞭さ
- 明確に書かれている
- 初心者へのやさしさ
- 82/100