fix(scheduler): base EMALong.update() on the decayed value — fixes #301

base-scheduler/src/main/java/org/apache/bifromq/basescheduler/EMALong.java

@@ -46,7 +46,12 @@ public void update(long newValue) {
         long now = nowSupplier.get();
         while (true) {
             State prev = state.get();
-            long newEma = (prev.ema == 0L) ? newValue : (long) Math.ceil(prev.ema * (1 - alpha) + newValue * alpha);
+            // Base the new sample on the *decayed* value, so update() and get() agree on the current
+            // EMA. Using the raw prev.ema here resurrects a stale congestion value after an idle period
+            // (get() reported it had decayed away) and re-arms the decay delay, which can pin the scheduler
+            // in permanent back-pressure.
+            long prevEma = decay(prev, now);
+            long newEma = (prevEma == 0L) ? newValue : (long) Math.ceil(prevEma * (1 - alpha) + newValue * alpha);
             State next = new State(newEma, now);
             if (state.compareAndSet(prev, next)) {
                 return;
@@ -55,8 +60,10 @@ public void update(long newValue) {
     }
 
     public long get() {
-        long now = nowSupplier.get();
-        State s = state.get();
+        return decay(state.get(), nowSupplier.get());
+    }
+
+    private long decay(State s, long now) {
         if (s.ema == 0L || s.lastTs == 0L) {
             return s.ema;
         }

base-scheduler/src/test/java/org/apache/bifromq/basescheduler/EMALongTest.java

@@ -113,4 +113,26 @@ void neverDecay() {
         // expected = 80
         assertEquals(ema.get(), 80);
     }
+
+    /**
+     * Regression: after an idle period a fresh sample must not resurrect the stale (decayed-away)
+     * congestion value. With the buggy base (undecayed prev.ema), update(0) after 60s idle would return
+     * 0.9 * 20s = 18s instead of ~32ms.
+     */
+    @Test
+    void updateAfterIdleUsesDecayedBase() {
+        EMALong ema = new EMALong(nowSupplier, 0.1, 0.9, 5_000_000_000L);
+        fakeTime.set(1L);
+        ema.update(20_000_000_000L); // congestion spike: ema = 20s
+        // idle for 60s beyond the decay delay: get() reports ≈ 20s * 0.9^60 ≈ 36ms
+        fakeTime.set(5_000_000_000L + 60_000_000_000L + 1L);
+        long decayedBefore = ema.get();
+        org.testng.Assert.assertTrue(decayedBefore < 1_000_000_000L,
+            "congestion should have decayed after 60s idle, but get()=" + decayedBefore);
+        // a fresh zero-latency sample must stay in the decayed range, not jump back to ~0.9 * 20s
+        ema.update(0L);
+        long after = ema.get();
+        org.testng.Assert.assertTrue(after < 1_000_000_000L,
+            "update() resurrected the stale value after idle: " + after);
+    }
 }