Time Series should use the IO.monotonic under the hood?

[floating-cat]

fs2 version: 3.2.11
Steps:

1. Run below code.
2. Change system time backward (e.g. in Linux sudo date --set='-10 seconds')

object Main extends IOApp.Simple {
  def run: IO[Unit] =
    TimeSeries.timePulled(
        Stream.iterateEval(1)(i => IO(i + 1)).metered(0.5.second),
        1.second,
        1.second
      )
      .evalMap(o => IO.println(o))
      .compile
      .drain
}

Current Result:
The code doesn't print any output because the time is backward and all values are dropped in these 10 seconds.

Expect Result:
All values including the tick signals are emitted.

Use cases:
In a client app, the user could change the time if their system time is not correct. Or the system clock is adjusted for some reason.

Reason:

In fs2 TimeSeries, the code use TimeStamped.reorderLocally to reoder the tickts.
But it use the TimeStamped.unsafeNow under the hood which use the System.currentTimeMillis() which is not monotonic.
If the time is backward,

• I use a small reorderOver value, my source steam values are dropped because the code is using TimeStamped.reorderLocally under the hood which would drop the values if values are not ordered and some combinators in the Time Series need the guarantee of the order too.
• I use a large reorderOver value, my original source steam values are reordered which breaks the code logic if I use the output stream directly

object TimeSeries {

  /** Stream of either time ticks (spaced by `tickPeriod`) or values from the source stream. */
  def apply[F[_]: Temporal, A](
      source: Stream[F, TimeStamped[A]],
      tickPeriod: FiniteDuration,
      reorderOver: FiniteDuration
  ): Stream[F, TimeStamped[Option[A]]] = {
    val src: Stream[F, TimeStamped[Option[A]]] = source.map(tsa => tsa.map(Some(_): Option[A]))
    val ticks: Stream[F, TimeStamped[Option[Nothing]]] =
      timeTicks(tickPeriod).map(tsu => tsu.map(_ => None))
    src.merge(ticks).through(TimeStamped.reorderLocally(reorderOver))
  }

  /** Stream of either time ticks (spaced by `tickPeriod`) or values from the source stream. */
  def timePulled[F[_]: Temporal, A](
      source: Stream[F, A],
      tickPeriod: FiniteDuration,
      reorderOver: FiniteDuration
  ): Stream[F, TimeStamped[Option[A]]] =
    apply(source.map(TimeStamped.unsafeNow), tickPeriod, reorderOver)

Should the Time Series use the IO.monotonic under the hood instead of the System.currentTimeMillis()?
Or maybe we should not use source values emitted from Time Series if any emitted values order/no loss are important?

[mpilquist]

@floating-cat Any thoughts on the approach in #2961? In essence, we provide either option to folks. We can't switch to monotonic time without significantly breaking client code.

[djspiewak]

As a note, monotonic is definitely intended for this type of case. It seems that the expectation when the code was written is that system time itself is also monotonic, and as the OP demonstrates, situations where this is not the case will in fact break.

I would recommend swapping to monotonic and adjusting it by the offset to realTime at stream start so as to get monotonicity without breaking anyone who expects it to be mostly system time ish. This also has the pleasing benefit of allowing this construct to be mocked using TestControl, which is impossible now if it's directly using currentTimeMillis.

[mpilquist]

@djspiewak I wrote the original, where the timestamp was used on a user interface. In the case of system time adjustments, we were okay with dropping data from reorderLocally, which used a small window size (typically a second, just to account for merging various sources that all used system time as their reference).

[djspiewak]

@mpilquist so the thought then is that the drift as system time adjusts is intentional and desirable? If that's the case then I'm aligned with your prior conclusion that we should just have both. 🙂 Swapping to realTime directly would be good though.

[floating-cat]

@mpilquist Hi, the approach in #2961 is good to me. The Swapping to realTime way is also fine to me.
Thanks for taking a look at this issue and creating a related PR so fast!