[2.x] fix: Discard a cancelled task's output backlog on cancel

A task that floods stdout could not be stopped promptly: cancelling it
(Ctrl+C from the thin client) stopped the task, but the server kept
draining the already-queued output to the client for several seconds,
because onCancellationRequest never discarded the pending backlog
(#9373).

onCancellationRequest now sets a per-channel discard flag and clears the
queued frames and buffered stdout (discardPending), and jsonRpcNotify
drops the cancelled task's systemOut/systemErr while the flag is set, so
output stops promptly. Only those two data channels are gated;
control-plane replies (including the cancel ack, enqueued after the
clear) and all notifications flow normally. It is a flag rather than a
one-shot clear because cancelAndShutdown only interrupts, so a CPU-bound
print loop keeps producing until it actually stops. The flag is reset at
the next prompt (prompt(ConsolePromptEvent), which fires per command),
so the following command's output is delivered normally.

The server-side heap growth under a sustained flood (the ~6.6GB in the
report) is a separate follow-up: bounding it safely first needs a
pre-existing forceFlush bug fixed (flushExecutor.shutdownNow() disables
output coalescing for the channel's life), which otherwise turns naive
back-pressure into a hang. Left out to keep this change surgical.

Refs #9373

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>

main/src/main/scala/sbt/internal/server/NetworkChannel.scala

@@ -92,6 +92,7 @@ final class NetworkChannel(
   private val pendingWrites = new LinkedBlockingQueue[(Array[Byte], Boolean)]()
   private val attached = new AtomicBoolean(false)
   private val alive = new AtomicBoolean(true)
+  private val isCanceled = new AtomicBoolean(false)
   private[sbt] def isInteractive = interactive.get
   private val interactive = new AtomicBoolean(false)
   private[sbt] def setInteractive(id: String, value: Boolean) = {
@@ -110,6 +111,10 @@ final class NetworkChannel(
     interactive.set(true)
     jsonRpcNotify(promptChannel, "")
   }
+  override def prompt(e: ConsolePromptEvent): Unit = {
+    isCanceled.set(false)
+    super.prompt(e)
+  }
   private[sbt] def write(byte: Byte) = inputBuffer.add(byte.toInt)
 
   private val terminalHolder = new AtomicReference[Terminal](Terminal.NullTerminal)
@@ -533,6 +538,8 @@ final class NetworkChannel(
                 StandardMain.exchange.currentExec.exists(_.source.exists(_.channelName == name)))
             ) {
               runningEngine.cancelAndShutdown()
+              isCanceled.set(true)
+              discardPending()
 
               respondResult(
                 ExecStatusEvent(
@@ -626,16 +633,19 @@ final class NetworkChannel(
   }
 
   /** Notify to Language Server's client. */
-  private[sbt] def jsonRpcNotify[A: JsonFormat](method: String, params: A): Unit = {
-    val m =
-      JsonRpcNotificationMessage("2.0", method, Option(Converter.toJson[A](params).get))
-    if (method != Serialization.systemOut) {
-      forceFlush()
-      log.debug(s"jsonRpcNotify: $m")
+  private[sbt] def jsonRpcNotify[A: JsonFormat](method: String, params: A): Unit =
+    if (isCanceled.get && NetworkChannel.isCanceledOutput(method))
+      ()
+    else {
+      val m =
+        JsonRpcNotificationMessage("2.0", method, Option(Converter.toJson[A](params).get))
+      if (method != Serialization.systemOut) {
+        forceFlush()
+        log.debug(s"jsonRpcNotify: $m")
+      }
+      val bytes = Serialization.serializeNotificationMessage(m)
+      publishBytes(bytes)
     }
-    val bytes = Serialization.serializeNotificationMessage(m)
-    publishBytes(bytes)
-  }
 
   /** Notify to Language Server's client. */
   private[sbt] def jsonRpcRequest[A: JsonFormat](id: String, method: String, params: A): Unit = {
@@ -676,6 +686,11 @@ final class NetworkChannel(
 
   import scala.jdk.CollectionConverters.*
   private val outputBuffer = new LinkedBlockingQueue[Byte]
+  private def discardPending(): Unit = {
+    pendingWrites.clear()
+    outputBuffer.synchronized(outputBuffer.clear())
+  }
+
   // Batches writes to the client at most once per 20ms to cut terminal flicker (see
   // CoalescingFlusher). forceFlush drains now while leaving the timer live.
   private val flusher =
@@ -943,6 +958,9 @@ final class NetworkChannel(
 }
 
 object NetworkChannel {
+  private[server] def isCanceledOutput(method: String): Boolean =
+    method == Serialization.systemOut || method == Serialization.systemErr
+
   private[sbt] def cancel(
       execID: Option[String],
       id: String,

main/src/test/scala/sbt/internal/server/NetworkChannelSpec.scala

@@ -0,0 +1,35 @@
+/*
+ * sbt
+ * Copyright 2023, Scala center
+ * Copyright 2011 - 2022, Lightbend, Inc.
+ * Copyright 2008 - 2010, Mark Harrah
+ * Licensed under Apache License 2.0 (see LICENSE)
+ */
+
+package sbt.internal.server
+
+import sbt.protocol.Serialization
+import verify.BasicTestSuite
+
+object NetworkChannelSpec extends BasicTestSuite:
+
+  test("only systemOut and systemErr are dropped while canceling") {
+    assert(NetworkChannel.isCanceledOutput(Serialization.systemOut))
+    assert(NetworkChannel.isCanceledOutput(Serialization.systemErr))
+  }
+
+  test("control-plane and flush methods are never dropped while canceling") {
+    val kept = Seq(
+      Serialization.systemOutFlush,
+      Serialization.systemErrFlush,
+      Serialization.readSystemIn,
+      Serialization.promptChannel,
+      "build/logMessage",
+      "sbt/exec",
+      "window/logMessage",
+      sbt.BasicCommandStrings.Shutdown,
+    )
+    kept.foreach(m => assert(!NetworkChannel.isCanceledOutput(m), s"must not drop: $m"))
+  }
+
+end NetworkChannelSpec

notes/2.0.0/cancel-discards-output.md

@@ -0,0 +1,8 @@
+### Cancelling a task discards its queued output
+
+Previously, cancelling a task that was producing a lot of output (Ctrl+C from the
+thin client) stopped the task, but the server kept draining the already-queued
+output to the client for several seconds, so the cancel appeared not to take
+effect. The cancelled task's buffered stdout/stderr is now discarded on cancel,
+so output stops promptly. Control-plane messages and the next command's output
+are unaffected.