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

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.

onCancellationRequest now sets a per-channel isCanceled 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.

Co-authored-by: Claude Fable 5 <noreply@anthropic.com>

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

@@ -91,6 +91,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) = {
@@ -109,6 +110,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)
@@ -532,6 +537,8 @@ final class NetworkChannel(
                 StandardMain.exchange.currentExec.exists(_.source.exists(_.channelName == name)))
             ) {
               runningEngine.cancelAndShutdown()
+              isCanceled.set(true)
+              discardPending()
 
               respondResult(
                 ExecStatusEvent(
@@ -625,16 +632,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 = {
@@ -675,6 +685,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 =
@@ -942,6 +957,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.