He-Pin opened a new issue, #3216:
URL: https://github.com/apache/pekko/issues/3216
### Motivation
`Source.queue` 的 `offer()` 返回的 Future 在元素被推送到下游之前就完成了（仅入队到
buffer），导致用户误以为元素已被流处理，但实际元素可能仍在 buffer 中等待。
### 当前代码行为
`QueueSource.scala` 中：
```scala
// QueueSource.scala:76-79, 131-138
def enqueueAndSuccess(elem: T): Unit = {
buffer.enqueue(elem)
offer.promise.success(QueueOfferResult.Enqueued) // 立即完成
...
}
```
在 buffered 模式（`maxBuffer > 0`）下，`offer.promise.success(Enqueued)` 在
`buffer.enqueue(...)` 后立即调用，此时元素尚未被 `push(out, elem)` 推送到下游。
仅 `maxBuffer == 0` 分支（137 行）在 `push(out, elem)` 后才完成 promise。
### 预期行为
`offer()` 的 Future 应该在元素实际被推送到下游流中时完成，而非仅入队到内部 buffer 时。至少应该在文档中明确说明此语义。
### 代码证据
-
`stream/src/main/scala/org/apache/pekko/stream/impl/QueueSource.scala:76-79` —
`enqueueAndSuccess` 方法
-
`stream/src/main/scala/org/apache/pekko/stream/impl/QueueSource.scala:131-138`
— buffered vs unbuffered 分支对比
### 复现方式
```scala
val queue = Source.queue[String](100, OverflowStrategy.backpressure)
.map(_ => Thread.sleep(1000)) // 模拟慢速下游
.to(Sink.ignore).run()
val f = queue.offer("test")
// f 几乎立即完成，但元素需要 1 秒才能通过 map
```
### 影响范围
- 模块：`pekko-stream`
- 影响依赖 `offer()` future 完成来判断消息已被处理的用户代码
- 在背压场景下可能导致用户错误地认为消息已送达
--
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]
