SessionState registry is unsynchronized but iterated from cluster and scheduler threads
- Dominant language
- Scala
- Stars
- 314
- Forks
- 187
- Avg merge
- 1d 21h
- Merged PRs (30d)
- 214
Description
### What happened?
`SessionState`'s registry is a plain unsynchronized `mutable.HashMap`, and it is iterated from several threads other than the websocket thread that mutates it.
```scala
// SessionState.scala:32
private val sessionIdToSessionState = new mutable.HashMap[String, SessionState]()
// :47 — hands out a live view, not a snapshot
def getAllSessionStates: Iterable[SessionState] = sessionIdToSessionState.values
```
`setState` / `removeState` are called from the websocket container's `@OnOpen` / `@OnClose`, while `getAllSessionStates` is iterated from at least three other places:
- `ClusterListener.scala:141` — a Pekko cluster-event callback, on the cluster dispatcher
- `Coordinator.scala:126`
- `RegionExecutionManager.scala:667`
`.values` is a live view over the map, so a session opening or closing while one of those broadcasts is in flight can throw `ConcurrentModificationException`, and in the worst case corrupt the map's internal state.
### How to reproduce?
Observed directly: while writing `ClusterListenerSpec`, a leftover listener iterating `getAllSessionStates` during a cluster membership event threw `ConcurrentModificationException` when a test registered a session concurrently. In production the same window is a client connecting or disconnecting while a cluster-status, execution-status or region-completion broadcast fans out — narrow, but entirely reachable on a busy server.
### Version/Branch
1.3.0-incubating-SNAPSHOT (main)
### Expected behavior
Either make the registry thread-safe (`TrieMap`, or a `ConcurrentHashMap` wrapper) or have `getAllSessionStates` return a snapshot:
```scala
def getAllSessionStates: Iterable[SessionState] = sessionIdToSessionState.synchronized {
sessionIdToSessionState.values.toList
}
```
A snapshot is the smaller change and is what the three call sites actually want — each of them fans a message out to whoever was connected at that moment.
### Additional context
Note that `removeState` also throws on an unknown session id (`sessionIdToSessionState(sId)` on line 43 before the `remove`), which makes a double-close fail rather than no-op. Worth considering in the same fix, though it is a separate defect.
Contributor guide
Assessment
This issue has not been assessed yet.