He-Pin opened a new issue, #3162:
URL: https://github.com/apache/pekko/issues/3162
### Summary
`UnfoldResourceSource` can invoke the user-provided `close` function twice
on the same resource when the first `close()` call throws a `NonFatal`
exception. This can corrupt or leak external resources that are not
idempotent-close-safe (file descriptors, reference-counted handles, distributed
locks, etc.).
### Symptom
A stream using `Source.unfoldResource` that completes (normally or via
downstream cancel) can end up calling `close(resource)` twice if the first
`close` invocation throws.
### Location
`stream/src/main/scala/org/apache/pekko/stream/impl/UnfoldResourceSource.scala`
In `closeStage()`:
```scala
private def closeStage(): Unit =
try {
close(resource) // close called FIRST
open = false // guard set AFTER close — skipped if close throws
completeStage()
} catch {
case NonFatal(ex) => failStage(ex) // `open` remains true here
}
override def postStop(): Unit = {
if (open) close(resource) // second close when open was never cleared
}
```
### Reproduction scenario
1. Build a stream that consumes from `Source.unfoldResource(create, read,
close)`.
2. Arrange for `close(resource)` to throw a `NonFatal` exception (e.g.
`IOException`).
3. Terminate the stream either via:
- downstream cancel (triggers `onDownstreamFinish` → `closeStage()`), or
- `read` returning `None` (normal completion → `closeStage()`).
4. Observe that `close` is invoked twice.
### Root cause
In `closeStage()`, the `open = false` guard is assigned AFTER
`close(resource)`. If `close` throws, the assignment is skipped, `open` remains
`true`, and `postStop()` (which the interpreter always calls after `failStage`)
invokes `close(resource)` again.
This is inconsistent with two other code paths in the same stage that
correctly clear the guard BEFORE calling `close`:
- Supervision.Stop branch in `onPull` (around line 63): sets `open = false`
then calls `close`.
- `restartState()` (around line 79): sets `open = false` then calls `close`.
The async counterpart (`UnfoldResourceSourceAsync`) does not have this
problem — it clears `maybeResource = OptionVal.none` immediately after calling
close, so `postStop` never double-closes.
### Impact
- User-provided `close` has no idempotency contract; double-close can
corrupt state, release external resources twice, or produce confusing error
logs.
- The second close's exception is swallowed silently by the interpreter's
`finalizeStage` try/catch around `postStop`, masking the bug in production.
### Proposed fix
Set `open = false` before calling `close(resource)` in `closeStage()`,
matching the pattern already used in the Supervision.Stop branch and
`restartState()`:
```scala
private def closeStage(): Unit = {
open = false
try {
close(resource)
completeStage()
} catch {
case NonFatal(ex) => failStage(ex)
}
}
```
A directional test should assert that `close` is invoked exactly once when
`close` throws during stream completion, covering both the downstream-cancel
path and the normal-completion path.
### Environment
- Apache Pekko streams, current `main`
- Affects `Source.unfoldResource` (sync variant only; async variant
unaffected)
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
