feat: add TimestampOrderChecker, out of order and duplicate timestamps

Windowed algorithms assume time moves forward. When it does not, nothing fails loudly: a window holds samples that never coexisted, a rate is computed over a negative interval, a duplicate is counted twice. Retries, buffers flushed after a reconnect, several producers writing into one topic and a clock stepped backwards all cause this, and the cheap defence is to check the order explicitly.

Each timestamp is compared with the previous one and classified as INCREASING, DUPLICATE or BACKWARDS. The checker only reports, it does not reorder or drop anything, so lastTimestamp is always the most recent sample, not the largest. It also keeps the worst backwards step, which is what you need to size a reordering buffer. O(1) time per sample and O(1) state; non-finite timestamps are refused.

Tests: the first timestamp is always accepted, an ordered stream has no violations, duplicates and backwards steps are counted separately, the worst step is the largest one and not the last one, the stream is not reordered, scan returns one verdict per timestamp, and reset restores the initial state.

Co-authored-by: Oleksandr Klymenko <19151554+alxkm@users.noreply.github.com>
Signed-off-by: alxkm <19151554+alxkm@users.noreply.github.com>

Author: alxkm

src/main/java/com/thealgorithms/streaming/TimestampOrderChecker.java

@@ -0,0 +1,203 @@
+package com.thealgorithms.streaming;
+
+/**
+ * Watches the timestamps of a stream for samples that arrive out of order or twice.
+ *
+ * <p>Every windowed algorithm assumes that time moves forward. When it does not, nothing crashes:
+ * the window simply holds samples that never coexisted, a rate is computed over a negative interval,
+ * and a duplicate counts twice in an average. Retries, buffers flushed after a reconnect, several
+ * producers writing into one topic and a clock stepped backwards all produce exactly that, and the
+ * only cheap defence is to check the order explicitly.
+ *
+ * <p>Each timestamp is compared with the one immediately before it and classified as
+ * {@link Order#INCREASING}, {@link Order#DUPLICATE} or {@link Order#BACKWARDS}. The checker reports,
+ * it does not reorder or drop anything, so {@link #lastTimestamp()} is always the most recent sample
+ * and not the largest one seen.
+ *
+ * <p>Besides the counters it keeps the worst backwards step, which is what a caller needs in order to
+ * size a reordering buffer: a stream whose worst violation is 20 milliseconds can be repaired by
+ * holding samples for 20 milliseconds, while one that jumps back an hour cannot be repaired at all.
+ *
+ * <h2>Usage</h2>
+ *
+ * <pre>{@code
+ * TimestampOrderChecker checker = new TimestampOrderChecker();
+ * for (double timestamp : timestamps) {
+ *     if (checker.accept(timestamp).isViolation()) {
+ *         quarantine(timestamp);
+ *     }
+ * }
+ * }</pre>
+ *
+ * <p>Each sample costs O(1) time and the checker keeps O(1) state. This class is not thread-safe.
+ *
+ * @see <a href="https://en.wikipedia.org/wiki/Monotonic_function">Monotonic function</a>
+ */
+public final class TimestampOrderChecker {
+
+    private double lastTimestamp;
+    private double worstBackwardsStep;
+    private boolean started;
+    private long count;
+    private long duplicateCount;
+    private long backwardsCount;
+    private Order lastOrder = Order.INCREASING;
+
+    /**
+     * Feeds the timestamp of one sample.
+     *
+     * @param timestamp when the sample arrived
+     * @return how this timestamp relates to the previous one; the first one is always
+     *     {@link Order#INCREASING}
+     * @throws IllegalArgumentException if {@code timestamp} is NaN or infinite
+     */
+    public Order accept(double timestamp) {
+        if (!Double.isFinite(timestamp)) {
+            throw new IllegalArgumentException("Timestamps must be finite, but was " + timestamp);
+        }
+
+        Order order = Order.INCREASING;
+        if (started) {
+            if (timestamp == lastTimestamp) {
+                order = Order.DUPLICATE;
+                duplicateCount++;
+            } else if (timestamp < lastTimestamp) {
+                order = Order.BACKWARDS;
+                backwardsCount++;
+                worstBackwardsStep = Math.max(worstBackwardsStep, lastTimestamp - timestamp);
+            }
+        }
+
+        lastTimestamp = timestamp;
+        started = true;
+        count++;
+        lastOrder = order;
+        return order;
+    }
+
+    /**
+     * Runs the checker over a whole series of timestamps.
+     *
+     * @param timestamps the timestamps to inspect
+     * @return a new array of the same length holding the verdict for every timestamp
+     * @throws IllegalArgumentException if a timestamp is NaN or infinite
+     * @throws NullPointerException if {@code timestamps} is {@code null}
+     */
+    public Order[] scan(double[] timestamps) {
+        Order[] verdicts = new Order[timestamps.length];
+        for (int i = 0; i < timestamps.length; i++) {
+            verdicts[i] = accept(timestamps[i]);
+        }
+        return verdicts;
+    }
+
+    /**
+     * Returns the verdict on the most recent timestamp.
+     *
+     * @return the last verdict, {@link Order#INCREASING} before the first timestamp
+     */
+    public Order lastOrder() {
+        return lastOrder;
+    }
+
+    /**
+     * Tells whether every timestamp so far moved forward.
+     *
+     * @return {@code true} if neither a duplicate nor a backwards step has been seen
+     */
+    public boolean isOrdered() {
+        return duplicateCount == 0 && backwardsCount == 0;
+    }
+
+    /**
+     * Returns the timestamp of the most recent sample, which is not necessarily the largest one seen.
+     *
+     * @return the last timestamp
+     * @throws IllegalStateException if no timestamp has been accepted yet
+     */
+    public double lastTimestamp() {
+        if (!started) {
+            throw new IllegalStateException("The checker has not seen any timestamp yet");
+        }
+        return lastTimestamp;
+    }
+
+    /**
+     * Returns the largest distance a timestamp ever jumped backwards, which is how long a reordering
+     * buffer would have to hold a sample to repair this stream.
+     *
+     * @return the worst backwards step, {@code 0} if the stream never went backwards
+     */
+    public double worstBackwardsStep() {
+        return worstBackwardsStep;
+    }
+
+    /**
+     * Returns how many timestamps have been inspected since the last reset.
+     *
+     * @return the sample count
+     */
+    public long count() {
+        return count;
+    }
+
+    /**
+     * Returns how many timestamps repeated the one before them.
+     *
+     * @return the duplicate count
+     */
+    public long duplicateCount() {
+        return duplicateCount;
+    }
+
+    /**
+     * Returns how many timestamps were smaller than the one before them.
+     *
+     * @return the backwards count
+     */
+    public long backwardsCount() {
+        return backwardsCount;
+    }
+
+    /**
+     * Forgets every timestamp seen so far.
+     */
+    public void reset() {
+        lastTimestamp = 0.0;
+        worstBackwardsStep = 0.0;
+        started = false;
+        count = 0;
+        duplicateCount = 0;
+        backwardsCount = 0;
+        lastOrder = Order.INCREASING;
+    }
+
+    @Override
+    public String toString() {
+        return "TimestampOrderChecker{samples=" + count + ", duplicates=" + duplicateCount + ", backwards=" + backwardsCount + ", worstStep=" + worstBackwardsStep + "}";
+    }
+
+    /**
+     * How a timestamp relates to the one before it.
+     */
+    public enum Order {
+
+        /** The timestamp moved forward, which is the only healthy case. */
+        INCREASING,
+
+        /** The timestamp repeats the previous one. */
+        DUPLICATE,
+
+        /** The timestamp is older than the previous one. */
+        BACKWARDS;
+
+        /**
+         * Tells whether this verdict reports a broken timeline.
+         *
+         * @return {@code true} for {@link #DUPLICATE} and {@link #BACKWARDS}
+         */
+        public boolean isViolation() {
+            return this != INCREASING;
+        }
+    }
+}

src/test/java/com/thealgorithms/streaming/TimestampOrderCheckerTest.java

@@ -0,0 +1,139 @@
+package com.thealgorithms.streaming;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertFalse;
+import static org.junit.jupiter.api.Assertions.assertThrows;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+import com.thealgorithms.streaming.TimestampOrderChecker.Order;
+import org.junit.jupiter.api.Assertions;
+import org.junit.jupiter.api.DisplayName;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.ValueSource;
+
+class TimestampOrderCheckerTest {
+
+    @ParameterizedTest
+    @ValueSource(doubles = {Double.NaN, Double.POSITIVE_INFINITY, Double.NEGATIVE_INFINITY})
+    void rejectsNonFiniteTimestamps(double timestamp) {
+        TimestampOrderChecker checker = new TimestampOrderChecker();
+        assertThrows(IllegalArgumentException.class, () -> checker.accept(timestamp));
+    }
+
+    @Test
+    void startsEmpty() {
+        TimestampOrderChecker checker = new TimestampOrderChecker();
+
+        assertEquals(0, checker.count());
+        assertEquals(Order.INCREASING, checker.lastOrder());
+        assertTrue(checker.isOrdered());
+        assertEquals(0.0, checker.worstBackwardsStep());
+        assertThrows(IllegalStateException.class, checker::lastTimestamp);
+    }
+
+    @Test
+    void theFirstTimestampIsAlwaysAccepted() {
+        TimestampOrderChecker checker = new TimestampOrderChecker();
+
+        assertEquals(Order.INCREASING, checker.accept(-500.0));
+        assertEquals(-500.0, checker.lastTimestamp());
+    }
+
+    @Test
+    void anOrderedStreamHasNoViolations() {
+        TimestampOrderChecker checker = new TimestampOrderChecker();
+
+        for (int i = 0; i < 20; i++) {
+            assertEquals(Order.INCREASING, checker.accept(i));
+        }
+        assertTrue(checker.isOrdered());
+        assertEquals(20, checker.count());
+    }
+
+    @Test
+    void reportsADuplicate() {
+        TimestampOrderChecker checker = new TimestampOrderChecker();
+        checker.accept(10.0);
+
+        assertEquals(Order.DUPLICATE, checker.accept(10.0));
+        assertEquals(1, checker.duplicateCount());
+        assertEquals(0, checker.backwardsCount());
+        assertFalse(checker.isOrdered());
+    }
+
+    @Test
+    void reportsAStepBackwards() {
+        TimestampOrderChecker checker = new TimestampOrderChecker();
+        checker.accept(10.0);
+
+        assertEquals(Order.BACKWARDS, checker.accept(7.5));
+        assertEquals(1, checker.backwardsCount());
+        assertEquals(2.5, checker.worstBackwardsStep());
+    }
+
+    @Test
+    @DisplayName("the worst backwards step is the largest one, not the last one")
+    void keepsTheWorstBackwardsStep() {
+        TimestampOrderChecker checker = new TimestampOrderChecker();
+
+        checker.scan(new double[] {100.0, 60.0, 61.0, 59.0, 200.0, 199.0});
+
+        assertEquals(3, checker.backwardsCount());
+        assertEquals(40.0, checker.worstBackwardsStep());
+    }
+
+    @Test
+    @DisplayName("the checker reports without reordering, so the last timestamp is the most recent one")
+    void doesNotReorderTheStream() {
+        TimestampOrderChecker checker = new TimestampOrderChecker();
+
+        checker.accept(100.0);
+        checker.accept(50.0);
+
+        assertEquals(50.0, checker.lastTimestamp());
+        assertEquals(Order.INCREASING, checker.accept(60.0), "the next sample is compared with the previous one, not with the maximum");
+    }
+
+    @Test
+    void scanReportsOneVerdictPerTimestamp() {
+        TimestampOrderChecker checker = new TimestampOrderChecker();
+
+        Order[] verdicts = checker.scan(new double[] {1.0, 2.0, 2.0, 1.5, 3.0});
+
+        Assertions.assertArrayEquals(new Order[] {Order.INCREASING, Order.INCREASING, Order.DUPLICATE, Order.BACKWARDS, Order.INCREASING}, verdicts);
+    }
+
+    @Test
+    void theVerdictKnowsWhetherItIsAViolation() {
+        assertFalse(Order.INCREASING.isViolation());
+        assertTrue(Order.DUPLICATE.isViolation());
+        assertTrue(Order.BACKWARDS.isViolation());
+    }
+
+    @Test
+    void resetRestoresTheInitialState() {
+        TimestampOrderChecker checker = new TimestampOrderChecker();
+        checker.scan(new double[] {5.0, 4.0, 4.0});
+
+        checker.reset();
+
+        assertEquals(0, checker.count());
+        assertEquals(0, checker.duplicateCount());
+        assertEquals(0, checker.backwardsCount());
+        assertEquals(0.0, checker.worstBackwardsStep());
+        assertTrue(checker.isOrdered());
+        assertThrows(IllegalStateException.class, checker::lastTimestamp);
+    }
+
+    @Test
+    void toStringMentionsTheState() {
+        TimestampOrderChecker checker = new TimestampOrderChecker();
+        checker.scan(new double[] {5.0, 4.0});
+
+        String text = checker.toString();
+
+        assertTrue(text.contains("TimestampOrderChecker"));
+        assertTrue(text.contains("backwards=1"));
+    }
+}