GROOVY-12459: Failed AsyncChannel producer never closes the channel so for await hangs and the exception is dropped

src/main/java/groovy/concurrent/BroadcastChannel.java

@@ -62,6 +62,10 @@
 public final class BroadcastChannel<T> {
 
     private final CopyOnWriteArrayList<AsyncChannel<T>> subscribers = new CopyOnWriteArrayList<>();
+    /** Guards {@link #closed} together with membership of {@link #subscribers}. */
+    private final Object lock = new Object();
+    /** Failure passed to {@link #close(Throwable)}, published before {@link #closed}. */
+    private volatile Throwable closeCause; // NOSONAR java:S3077 — published as a reference; the Throwable is not mutated
     private volatile boolean closed;
 
     private BroadcastChannel() { }
@@ -94,9 +98,11 @@ public AsyncChannel<T> subscribe() {
      * @return a new subscriber channel
      */
     public AsyncChannel<T> subscribe(int bufferSize) {
-        if (closed) throw new ChannelClosedException("BroadcastChannel is closed");
         AsyncChannel<T> sub = AsyncChannel.create(bufferSize);
-        subscribers.add(sub);
+        synchronized (lock) {
+            if (closed) throw closedNow();
+            subscribers.add(sub);
+        }
         return sub;
     }
 
@@ -105,11 +111,12 @@ public AsyncChannel<T> subscribe(int bufferSize) {
      *
      * @param value the value to broadcast
      * @return an Awaitable that completes when all subscribers have accepted the value
-     * @throws ChannelClosedException if the broadcast channel is closed
+     * @throws ChannelClosedException if the broadcast channel is closed.
+     *         The cause is the failure passed to {@link #close(Throwable)}, when there is one.
      */
     public Awaitable<Void> send(T value) {
         Objects.requireNonNull(value, "value must not be null");
-        if (closed) throw new ChannelClosedException("BroadcastChannel is closed");
+        if (closed) throw closedNow();
         CompletableFuture<?>[] futures = subscribers.stream()
                 .map(sub -> {
                     try {
@@ -124,12 +131,56 @@ public Awaitable<Void> send(T value) {
 
     /**
      * Closes this broadcast channel and all subscriber channels.
+     * Idempotent. A later {@link #close(Throwable)} does not add a failure.
      */
     public void close() {
-        closed = true;
+        close(null);
+    }
+
+    /**
+     * Closes this broadcast channel because {@code cause} failed the producer.
+     * <p>
+     * Values already accepted by a subscriber remain receivable. Iteration
+     * on each subscriber throws {@code cause}. {@code null} is a normal close,
+     * the same as {@link #close()}. Returns {@code true} when this call closes
+     * the broadcast. The first close wins, so a later call returns {@code false}
+     * and leaves the retained failure unchanged. A subscriber that arrives
+     * afterwards fails at once with {@link ChannelClosedException} carrying
+     * this cause.
+     *
+     * @param cause the failure each subscriber should surface, or {@code null}
+     * @return {@code true} if this call closed the broadcast
+     * @since 6.0.1
+     */
+    public boolean close(Throwable cause) {
+        synchronized (lock) {
+            if (closed) {
+                return false;
+            }
+            closeCause = cause;
+            closed = true;
+        }
         for (AsyncChannel<T> sub : subscribers) {
-            sub.close();
+            sub.close(cause);
         }
+        return true;
+    }
+
+    /**
+     * Returns the failure passed to {@link #close(Throwable)}, or {@code null}
+     * when this broadcast is open or was closed normally.
+     *
+     * @return the retained failure, or {@code null}
+     * @since 6.0.1
+     */
+    public Throwable getCloseCause() {
+        return closeCause;
+    }
+
+    private ChannelClosedException closedNow() {
+        Throwable cause = closeCause;
+        if (cause == null) return new ChannelClosedException("BroadcastChannel is closed");
+        return new ChannelClosedException("BroadcastChannel is closed", cause);
     }
 
     /**
@@ -162,8 +213,9 @@ public int getSubscriberCount() {
      *       broadcast send when no demand exists (sender-side backpressure).</li>
      *   <li>Cancellation: closes this subscriber's channel and removes it
      *       from the broadcast's subscriber set.</li>
-     *   <li>Completion: signals {@code onComplete} when the broadcast channel
-     *       is closed and the per-subscriber buffer drained.</li>
+     *   <li>Completion: signals {@code onComplete} when the subscriber channel
+     *       is closed normally and drained. {@link #close(Throwable)} signals
+     *       {@code onError} with that failure instead.</li>
      * </ul>
      * <p>
      * <b>Backpressure policy (important).</b> This bridge uses lossless,
@@ -302,8 +354,7 @@ private void drain() {
                     // If the upstream channel has closed and drained, emit
                     // onComplete even when demand == 0.
                     if (channel.isClosed() && channel.getBufferedSize() == 0) {
-                        subscriber.onComplete();
-                        cancel();
+                        completeOrFail(null);
                         return;
                     }
                     if (demand.get() == 0) {
@@ -322,10 +373,7 @@ private void drain() {
                     try {
                         item = AsyncSupport.await(channel.receive());
                     } catch (ChannelClosedException e) {
-                        if (!cancelled.get()) {
-                            subscriber.onComplete();
-                            cancel();
-                        }
+                        completeOrFail(e.getCause());
                         return;
                     }
                     if (cancelled.get()) return;
@@ -346,5 +394,21 @@ private void drain() {
                 }
             }
         }
+
+        /** A retained failure fails the subscriber. A normal close completes it. */
+        private void completeOrFail(Throwable cause) {
+            if (cancelled.get()) {
+                return;
+            }
+            if (cause == null) {
+                cause = channel.getCloseCause();
+            }
+            if (cause != null) {
+                subscriber.onError(cause);
+            } else {
+                subscriber.onComplete();
+            }
+            cancel();
+        }
     }
 }

src/main/java/groovy/concurrent/ChannelClosedException.java

@@ -30,14 +30,16 @@
  *       or during the send attempt.  Pending senders that were waiting for buffer
  *       space when the channel closed also receive this exception.</li>
  *   <li>{@link AsyncChannel#receive() receive()} — the channel was closed and all
- *       buffered values have been drained.  Note that values buffered before
- *       closure are still delivered normally; only once the buffer is exhausted
- *       does this exception appear.</li>
+ *       buffered values have been drained. Values buffered before closure are
+ *       still delivered. A failure passed to {@link AsyncChannel#close(Throwable)}
+ *       is this exception's {@linkplain Throwable#getCause() cause}, not a
+ *       different exception type. A hand-written loop must read that cause.</li>
  * </ul>
  * <p>
- * When used with {@code for await}, the loop infrastructure translates
- * {@code ChannelClosedException} into a clean end-of-stream signal (i.e.,
- * the loop exits normally rather than propagating the exception):
+ * When used with {@code for await}, a plain {@code for}, {@code each},
+ * {@code collect} or {@code toList}, a close with no cause ends iteration.
+ * A close that carries a cause throws that cause after the values already
+ * sent. The code that reads the channel writes the {@code try}/{@code catch}:
  * <pre>
  * def ch = AsyncChannel.create()
  * // ... producer sends values, then calls ch.close()
@@ -46,6 +48,8 @@
  * }
  * // loop exits cleanly after the channel is closed and drained
  * </pre>
+ * A Java enhanced {@code for} calls {@code Iterator.hasNext()}, which may
+ * throw a checked exception that {@code Iterator} does not declare.
  *
  * @see AsyncChannel#send(Object)
  * @see AsyncChannel#receive()

src/main/java/groovy/concurrent/ChannelSelect.java

@@ -332,8 +332,16 @@ public ChannelSelect random() {
      * exception, add a timer offer from {@link #after(long)} or receive
      * from a timer channel from {@link AsyncChannel#after(long)}.
      * <p>
-     * If every offer's channel is closed, the result fails with
-     * {@link ChannelClosedException}. If every offer is guarded off (see
+     * If every offer's channel is closed normally, the result fails with
+     * {@link ChannelClosedException}. A channel closed with a failure still
+     * delivers a value that is already buffered. The failure fails the
+     * select only when that offer's next operation is the close. Which of
+     * a buffered value and such a failure is met first follows registration
+     * order: list order here, rotation for {@link #fair()}, and a shuffle
+     * for {@link #random()}. A transfer that has already committed stays
+     * the result: the claim admits one outcome, so that value is not
+     * discarded when a close races it.
+     * If every offer is guarded off (see
      * {@link Offer#when(BooleanSupplier)}) there is nothing to wait for and
      * the result fails with {@link IllegalStateException}.
      * <p>
@@ -441,8 +449,14 @@ private Awaitable<Result> start(int[] order) {
                     } else {
                         resend(ch, value);
                     }
-                } else if (error instanceof ChannelClosedException && closedCount.incrementAndGet() == count) {
-                    winner.completeExceptionally(new ChannelClosedException("all channels are closed"));
+                } else if (error instanceof ChannelClosedException closed) {
+                    Throwable cause = closed.getCause();
+                    if (cause != null) {
+                        failIfUnclaimed(winner, branches, timers, cause);
+                    } else if (closedCount.incrementAndGet() == count) {
+                        failIfUnclaimed(winner, branches, timers,
+                                new ChannelClosedException("all channels are closed"));
+                    }
                 }
             });
         }
@@ -500,6 +514,19 @@ private boolean isEnabled(int index, boolean[] enabled) {
         return (enabled == null || enabled[index]) && offers.get(index).isEnabled();
     }
 
+    /**
+     * Fails {@code winner} with {@code cause} only when no offer has committed.
+     * The claim is the same gate {@link Winner#cancel} uses, so a transfer
+     * that already holds it stays the result and its value is not discarded.
+     * Losers are withdrawn before the caller observes the failure.
+     */
+    private static void failIfUnclaimed(Winner winner, Awaitable<?>[] branches,
+                                        AsyncChannel<?>[] timers, Throwable cause) {
+        if (!winner.claim.tryCommitCancel()) return;
+        withdraw(branches, timers);
+        winner.completeExceptionally(cause);
+    }
+
     /**
      * Withdraws every registered branch, and closes every timer armed for
      * the call so its scheduled firing is cancelled. Branches go first: a

src/main/java/org/apache/groovy/runtime/async/AsyncSupport.java

@@ -546,6 +546,14 @@ public static Throwable unwrap(Throwable t) {
         return t;
     }
 
+    /**
+     * Rethrows {@code failure} as itself, including a checked exception.
+     * Kept in this package so the pure-Java concurrent jar stays self-contained.
+     */
+    static void rethrow(Throwable failure) {
+        sneakyThrow(failure);
+    }
+
     @SuppressWarnings("unchecked")
     private static <T extends Throwable> void sneakyThrow(Throwable t) throws T {
         throw (T) t;

src/main/java/org/apache/groovy/runtime/async/DefaultAsyncChannel.java

@@ -71,6 +71,8 @@ public final class DefaultAsyncChannel<T> implements AsyncChannel<T> {
     private final Deque<PendingOp<T>> waitingReceivers = new ConcurrentLinkedDeque<>();
     private final int capacity;
     private volatile boolean closed;
+    /** Failure passed to {@link #close(Throwable)}, published before {@link #closed}. */
+    private volatile Throwable closeCause; // NOSONAR java:S3077 — published as a reference; the Throwable is not mutated
     /**
      * Run once, outside the lock, by the call that closes the channel; {@code null} for none.
      * {@link AtomicReference} rather than {@code volatile}: java:S3077 flags a volatile
@@ -286,9 +288,27 @@ private Awaitable<T> receive(SelectClaim claim) {
 
     @Override
     public boolean close() {
+        return finishClose(null);
+    }
+
+    @Override
+    public Throwable getCloseCause() {
+        return closeCause;
+    }
+
+    /** Retains {@code cause} and closes. Does not call the interface default. */
+    @Override
+    public boolean close(Throwable cause) {
+        return finishClose(cause);
+    }
+
+    private boolean finishClose(Throwable cause) {
         lock.lock();
         try {
-            if (closed) return false;
+            if (closed) {
+                return false;
+            }
+            closeCause = cause;
             closed = true;
             finishCloseLocked();
         } finally {
@@ -344,7 +364,13 @@ private void runCloseHook() {
     /**
      * Returns a blocking iterator that receives values until the channel
      * is closed and drained. Each {@code next()} call blocks until a value
-     * is available. {@link ChannelClosedException} signals end-of-iteration.
+     * is available. A {@link ChannelClosedException} with no cause ends
+     * iteration. A failure retained by {@link #close(Throwable)} is that
+     * exception's cause and is thrown here, once, for the caller to catch.
+     * {@code for}, {@code each}, {@code collect} and {@code toList} all use
+     * this iterator. A Java enhanced {@code for} calls {@code hasNext()},
+     * which may throw a checked exception that {@link Iterator} does not
+     * declare.
      */
     @Override
     public Iterator<T> iterator() {
@@ -364,6 +390,10 @@ public boolean hasNext() {
                 return true;
             } catch (ChannelClosedException e) {
                 done = true;
+                Throwable failure = e.getCause();
+                if (failure != null) {
+                    AsyncSupport.rethrow(failure);
+                }
                 return false;
             }
         }
@@ -510,12 +540,21 @@ private void refillBufferFromWaitingSenders() {
         }
     }
 
-    private static ChannelClosedException closedForSend() {
-        return new ChannelClosedException("channel is closed for send");
+    private ChannelClosedException closedForSend() {
+        return closed("channel is closed for send");
+    }
+
+    private ChannelClosedException closedForReceive() {
+        return closed("channel is closed");
     }
 
-    private static ChannelClosedException closedForReceive() {
-        return new ChannelClosedException("channel is closed");
+    /** One terminal signal. A retained failure rides on the cause. */
+    private ChannelClosedException closed(String message) {
+        Throwable cause = closeCause;
+        if (cause == null) {
+            return new ChannelClosedException(message);
+        }
+        return new ChannelClosedException(message, cause);
     }
 
     /**