Stop searching for work when stolen timers filled the local queue

core/jvm-native/src/main/scala/cats/effect/unsafe/LocalQueue.scala

@@ -331,13 +331,7 @@ private final class LocalQueue extends LocalQueuePadding {
     val tl = tail
 
     while (true) {
-      // A load of the head of the queue using `acquire` semantics.
-      val hd = Head.updater.get(this)
-      val steal = msb(hd)
-
-      // Check the usable capacity of the queue.
-      val len = unsignedShortSubtraction(tl, steal)
-      if (len <= LocalQueueCapacityMinusBatch) {
+      if (hasCapacityForBatch()) {
         // It is safe to transfer the fibers from the batch to the queue.
         val startPos = tl - 1
         var i = 1
@@ -368,7 +362,8 @@ private final class LocalQueue extends LocalQueuePadding {
         return fiber
       }
 
-      // Not enough usable capacity, which means there is an ongoing steal
+      // Not enough usable capacity. Callers guarantee that the queue has spare
+      // capacity to hold the batch, which means there is an ongoing steal
       // operation. Spin until it completes.
     }
 
@@ -379,6 +374,20 @@ private final class LocalQueue extends LocalQueuePadding {
     null
   }
 
+  /**
+   * Checks whether this queue has enough usable capacity to hold a batch of fibers.
+   *
+   * @note
+   *   Can '''only''' be correctly called by the owner [[WorkerThread]].
+   *
+   * @return
+   *   `true` if a batch of fibers can be enqueued on this local queue, `false` otherwise
+   */
+  def hasCapacityForBatch(): Boolean = {
+    val hd = Head.updater.get(this)
+    unsignedShortSubtraction(tail, msb(hd)) <= LocalQueueCapacityMinusBatch
+  }
+
   /**
    * Dequeues a fiber from the head of the local queue.
    *

core/jvm-native/src/main/scala/cats/effect/unsafe/WorkStealingThreadPool.scala

@@ -203,6 +203,9 @@ private[effect] final class WorkStealingThreadPool[P <: AnyRef](
    * worker threads starting at a random index. If the stealing attempt was unsuccessful, this
    * method falls back to checking the external queue.
    *
+   * Stealing from external queue is skipped if the calling worker thread's own local queue is
+   * too full to take a batch of fibers from the external queue.
+   *
    * @param dest
    *   the index of the worker thread attempting to steal work from other worker threads (used
    *   to avoid stealing from its own local queue)
@@ -239,7 +242,16 @@ private[effect] final class WorkStealingThreadPool[P <: AnyRef](
     }
 
     // The worker thread could not steal any work. Fall back to checking the
-    // external queue.
+    // external queue, but only if the local queue can hold a batch of fibers.
+    // Expired timers stolen just before this call may have resumed enough
+    // fibers onto the local queue to leave no room for a batch, in which case
+    // `enqueueBatch` would spin forever, as it expects to be called only when
+    // there is capacity for the batch.
+    // See https://github.com/typelevel/cats-effect/issues/4674.
+    if (!destQueue.hasCapacityForBatch()) {
+      return null
+    }
+
     val element = externalQueue.poll(random)
     if (element.isInstanceOf[Array[Runnable]]) {
       val batch = element.asInstanceOf[Array[Runnable]]