Fix bug in fixedRate which fails to dampen ticks when multiple period…

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

@@ -3125,16 +3125,18 @@ object Stream extends StreamLowPriority {
     else
       Stream.eval(Temporal[F].monotonic).flatMap { now =>
         val next = lastAwakeAt + period
-        val step =
-          if (next > now) Stream.sleep(next - now)
-          else {
-            (now.toNanos - lastAwakeAt.toNanos - 1) / period.toNanos match {
+        if (next > now)
+          Stream.sleep(next - now) ++ fixedRate_(period, next, dampen)
+        else {
+          val ticks = (now.toNanos - lastAwakeAt.toNanos - 1) / period.toNanos
+          val step =
+            ticks match {
               case count if count < 0            => Stream.empty
               case count if count == 0 || dampen => Stream.emit(())
               case count                         => Stream.emit(()).repeatN(count)
             }
-          }
-        step ++ fixedRate_(period, next, dampen)
+          step ++ fixedRate_(period, lastAwakeAt + (period * ticks), dampen)
+        }
       }
 
   private[fs2] final class PartiallyAppliedFromOption[F[_]](
@@ -4305,22 +4307,26 @@ object Stream extends StreamLowPriority {
       private val underlying: Pull[F, O, Unit]
   )(implicit compiler: Compiler[F, G]) {
 
-    /** Compiles this stream in to a value of the target effect type `F` and
+    /** Compiles this stream to a count of the elements in the target effect type `G`.
+      */
+    def count: G[Long] = foldChunks(0L)((acc, chunk) => acc + chunk.size)
+
+    /** Compiles this stream in to a value of the target effect type `G` and
       * discards any output values of the stream.
       *
       * To access the output values of the stream, use one of the other compilation methods --
       * e.g., [[fold]], [[toVector]], etc.
       */
     def drain: G[Unit] = foldChunks(())((_, _) => ())
 
-    /** Compiles this stream in to a value of the target effect type `F` by folding
+    /** Compiles this stream in to a value of the target effect type `G` by folding
       * the output values together, starting with the provided `init` and combining the
       * current value with each output value.
       */
     def fold[B](init: B)(f: (B, O) => B): G[B] =
       foldChunks(init)((acc, c) => c.foldLeft(acc)(f))
 
-    /** Compiles this stream in to a value of the target effect type `F` by folding
+    /** Compiles this stream in to a value of the target effect type `G` by folding
       * the output chunks together, starting with the provided `init` and combining the
       * current value with each output chunk.
       *
@@ -4355,7 +4361,7 @@ object Stream extends StreamLowPriority {
     def foldSemigroup(implicit O: Semigroup[O]): G[Option[O]] =
       fold(Option.empty[O])((acc, o) => acc.map(O.combine(_, o)).orElse(Some(o)))
 
-    /** Compiles this stream in to a value of the target effect type `F`,
+    /** Compiles this stream in to a value of the target effect type `G`,
       * returning `None` if the stream emitted no values and returning the
       * last value emitted wrapped in `Some` if values were emitted.
       *
@@ -4371,7 +4377,7 @@ object Stream extends StreamLowPriority {
     def last: G[Option[O]] =
       foldChunks(Option.empty[O])((acc, c) => c.last.orElse(acc))
 
-    /** Compiles this stream in to a value of the target effect type `F`,
+    /** Compiles this stream in to a value of the target effect type `G`,
       * raising a `NoSuchElementException` if the stream emitted no values
       * and returning the last value emitted otherwise.
       *
@@ -4390,7 +4396,7 @@ object Stream extends StreamLowPriority {
       last.flatMap(_.fold(G.raiseError(new NoSuchElementException): G[O])(G.pure))
 
     /** Gives access to the whole compilation api, where the result is
-      * expressed as a `cats.effect.Resource`, instead of bare `F`.
+      * expressed as a `cats.effect.Resource`, instead of bare `G`.
       *
       * {{{
       *  import fs2._
@@ -4491,7 +4497,7 @@ object Stream extends StreamLowPriority {
     def string(implicit ev: O <:< String): G[String] =
       new Stream(underlying).asInstanceOf[Stream[F, String]].compile.to(Collector.string)
 
-    /** Compiles this stream into a value of the target effect type `F` by collecting
+    /** Compiles this stream into a value of the target effect type `G` by collecting
       * all of the output values in a collection.
       *
       * Collection building is done via an explicitly passed `Collector`.
@@ -4526,7 +4532,7 @@ object Stream extends StreamLowPriority {
       } yield builder.result
     }
 
-    /** Compiles this stream in to a value of the target effect type `F` by logging
+    /** Compiles this stream in to a value of the target effect type `G` by logging
       * the output values to a `List`. Equivalent to `to[List]`.
       *
       * When this method has returned, the stream has not begun execution -- this method simply
@@ -4540,7 +4546,7 @@ object Stream extends StreamLowPriority {
       */
     def toList: G[List[O]] = to(List)
 
-    /** Compiles this stream in to a value of the target effect type `F` by logging
+    /** Compiles this stream in to a value of the target effect type `G` by logging
       * the output values to a `Vector`. Equivalent to `to[Vector]`.
       *
       * When this method has returned, the stream has not begun execution -- this method simply

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

@@ -89,6 +89,22 @@ class StreamCombinatorsSuite extends Fs2Suite {
         .toVector
         .assertEquals(input.compile.toVector)
     }
+
+    test("dampening") {
+      val count = 10L
+      val period = 10.millis
+      Stream
+        .awakeEvery[IO](period)
+        .take(count)
+        .mapAccumulate(0L)((acc, o) => (acc + 1, o))
+        .evalMap { case (i, o) => IO.sleep(if (i == 2) 5 * period else 0.seconds).as(o) }
+        .compile
+        .toVector
+        .map { v =>
+          val elapsed = v.last - v.head
+          assert(elapsed > count * period)
+        }
+    }
   }
 
   group("buffer") {