aws-amplify / aws-amplify/amplify-android
Events: cancelling one subscriber can silently kill a shared WebSocket connection for others
- Dominant language
- Java
- Stars
- 287
- Forks
- 132
- Avg merge
- 2d 2h
- Merged PRs (30d)
- 45
Description
### Summary
In `aws-sdk-appsync-events`, the shared WebSocket connection attempt is launched in the coroutine of whichever caller happens to arrive first. If that caller is cancelled while others are waiting on the same attempt, they are cancelled too — and because the failure arrives as a `CancellationException`, it is indistinguishable from those callers being cancelled themselves. No error surfaces.
I found this while writing a provider with the same shape elsewhere, and traced the pattern back here.
### Where
`appsync/aws-sdk-appsync-events/src/main/java/com/amazonaws/sdk/appsync/events/EventsWebSocketProvider.kt`
```kotlin
private suspend fun getConnectedWebSocketResult(): Result = coroutineScope {
...
val newDeferredInProgressConnection = async { attemptConnection() }
connectionInProgressReference.set(newDeferredInProgressConnection)
val connectionResult = newDeferredInProgressConnection.await()
...
}
```
### Mechanism
`async { }` inherits the `coroutineScope { }` of the first caller, so the `Deferred` is that caller's child. It is then published to `connectionInProgressReference`, where later callers join it:
```kotlin
val deferredInProgressConnection = connectionInProgressReference.get()
if (deferredInProgressConnection != null && !deferredInProgressConnection.isCompleted) {
return@coroutineScope deferredInProgressConnection.await()
}
```
So when the first caller is cancelled, its scope cancels the `async` child, and every joined caller's `await()` throws `CancellationException`. Structured concurrency then treats that as *the joiner* being cancelled, so the joiner stops without raising anything a caller can catch. The connection they were waiting for simply never arrives, and nothing explains why.
The attempt belongs to the provider, not to whichever caller reached it first — the lifetime is the thing that is wrong here, not the locking.
### Why this is reachable in normal use
`getConnectedWebSocket()` is called from inside `onStart` on the flow returned by `EventsWebSocketClient.subscribe(...)`:
```kotlin
return createSubscriptionEventDataFlow(subscriptionHolder)
.onStart {
val newWebSocket = eventsWebSocketProvider.getConnectedWebSocket()
...
}
```
That places the connection attempt in the **collector's** coroutine, and a collector ending is entirely routine: `take(n)`, a timeout, `launchIn(viewModelScope)` on a screen that closes, or a `collect` inside a job that gets cancelled. Two subscriptions starting at roughly the same time are enough — if the first collector goes away during the handshake, the second silently gets nothing.
This is only a problem while an attempt is in flight, so it needs concurrent subscribers to bite. It is not a persistent broken state: a cancelled `Deferred` reports `isCompleted == true`, so the `!isCompleted` guard sends the *next* caller down the path of starting a fresh attempt. It is the callers already waiting who lose.
### Impact
- A subscription flow that stops before emitting, with no error delivered to the collector.
- Nothing distinguishes it from ordinary cancellation, so it is close to undebuggable from an app's perspective — there is no exception, no log, and the provider recovers by the next call.
- More likely under exactly the conditions apps create: several subscriptions started together during startup or a screen transition.
### Why the current tests do not catch it
`EventsWebSocketProviderTest.multiple calls return same instance when not closed` launches ten concurrent calls and `awaitAll()`s them. Nothing is ever cancelled, so the cross-caller cancellation path is never exercised. A concurrency test built on `awaitAll` structurally cannot catch this — every participant runs to completion by construction.
### Suggested fix
Give the provider its own scope so the attempt outlives any individual caller:
```kotlin
private val scope = CoroutineScope(ioDispatcher + SupervisorJob())
...
val newDeferredInProgressConnection = scope.async { attemptConnection() }
```
One wrinkle worth flagging rather than glossing over: `EventsWebSocketProvider` currently has no teardown, so a provider-owned scope needs somewhere to be cancelled — presumably from `EventsWebSocketClient.disconnect(...)`, which is the existing lifecycle boundary. That is the part of the change that needs a decision, not the `async` itself.
Adjacent, lower priority: `async` and its `await()` are both inside `mutex.withLock`, so the mutex is held for the whole network connect. The pre-lock `connectionInProgressReference` check covers the common case, but a caller arriving before that reference is published blocks on the mutex for the full connect rather than joining the attempt.
### Suggested regression test
Cancel the first caller specifically, rather than letting every participant finish:
```kotlin
@Test
fun `cancelling the first caller does not cancel the attempt others joined`() = runTest {
val gate = CompletableDeferred()
every { anyConstructed().isClosed } returns false
coEvery { anyConstructed().connect() } coAnswers { gate.await() }
val first = launch { provider.getConnectedWebSocket() }
val second = async { provider.getConnectedWebSocket() }
runCurrent()
first.cancel()
gate.complete(Unit)
// Fails before the fix: second is cancelled by first's cancellation.
second.await()
coVerify(exactly = 1) { anyConstructed().connect() }
}
```
### Environment
Observed by reading `main`; the code is unchanged since the module was added. Not reproduced against a live endpoint — the analysis is from the source and the call sites above.
Contributor guide
Research direction
Start with appsync/aws-sdk-appsync-events/src/main/java/com/amazonaws/sdk/appsync/events/EventsWebSocketProvider.kt and the EventsWebSocketClient.disconnect(...) lifecycle entry point. Add the cancellation scenario to EventsWebSocketProviderTest, then verify that cancelling the first caller does not cancel the shared attempt, that only one connection is made, and that provider teardown is covered.
Written by the indexing model from the issue text.
Assessment
- Tech stack
- kotlin
- Domain
- mobile-dev, networking
- Issue type
- Bug
- Difficulty
- 4/5
- Estimated time
- 3-5 days
- Activity status
- Active
- Clarity
- Mostly clear
- Newbie friendliness
- 58/100