fix rechunkRandomly with large ending chunk

core/shared/src/main/scala/fs2/Stream.scala

@@ -2383,23 +2383,41 @@ final class Stream[+F[_], +O] private[fs2] (private[fs2] val underlying: Pull[F,
 
       def nextSize(sourceSize: Int): Int = (factor * sourceSize).toInt
 
-      def go(acc: Chunk[O], sizeOpt: Int, s: Pull[F2, O, Unit]): Pull[F2, O, Unit] =
-        s.uncons.flatMap {
-          case Some((hd, tl)) =>
-            val size = if (sizeOpt > 0) sizeOpt else nextSize(hd.size)
-            if (acc.size < size)
-              go(acc ++ hd, size, tl)
-            else if (acc.size == size)
-              Pull.output(acc) >> go(hd, size, tl)
-            else {
-              val (out, rem) = acc.splitAt(size - 1)
-              Pull.output(out) >> go(rem ++ hd, -1, tl)
+      def go(
+          acc: Chunk[O],
+          sizeOpt: Int,
+          lastChunkSize: Int,
+          s: Pull[F2, O, Unit]
+      ): Pull[F2, O, Unit] = {
+
+        val size = if (sizeOpt > 0) sizeOpt else nextSize(lastChunkSize)
+
+        if (acc.size < size)
+          s.uncons.flatMap {
+            case None => Pull.output(acc)
+            case Some((hd, tl)) =>
+              go(acc ++ hd, size, hd.size, tl)
+          }
+        else if (acc.size == size)
+          Pull.output(acc) >>
+            s.uncons.flatMap {
+              case None => Pull.done
+              case Some((hd, tl)) =>
+                go(hd, size, hd.size, tl)
             }
-          case None =>
-            Pull.output(acc)
+        else {
+          val (out, rem) = acc.splitAt(size - 1)
+          Pull.output(out) >> go(rem, -1, lastChunkSize, s)
+
         }
 
-      go(Chunk.empty, -1, underlying).stream
+      }
+
+      underlying.uncons.flatMap {
+        case None => Pull.done
+        case Some((hd, tl)) =>
+          go(hd, -1, hd.size, tl)
+      }.stream
     }
 
   /** Rechunks the stream such that output chunks are within [inputChunk.size * minFactor, inputChunk.size * maxFactor].

core/shared/src/test/scala/fs2/StreamCombinatorsSuite.scala

@@ -1354,6 +1354,16 @@ class StreamCombinatorsSuite extends Fs2Suite {
         }
       }
     }
+
+    test("rechunkRandomlyWithSeed should correclty rechunk big chunks at the end of a stream") {
+      val chunks = Stream
+        .chunk(Chunk.seq(List.fill(5000)(1)))
+        .rechunkRandomlyWithSeed(0.01, 0.1)(1L)
+        .chunks
+        .compile
+        .toList
+      assert(chunks.forall(_.size < 500))
+    }
   }
 
   group("rechunkRandomly") {