fix: futures sometimes being wrongly tagged as unloaded

Author: ishland

src/main/java/com/ishland/flowsched/scheduler/ItemHolder.java

@@ -263,7 +263,19 @@ public boolean setStatus(ItemStatus<K, V, Ctx> status, boolean isCancellation) {
 //                }
 
                 this.status = status;
-                flushUnloadedStatus(prevStatus, true);
+
+                // reinit higher futures because downgraded, and deinit futures higher than target status
+                synchronized (this.futures) {
+                    final ItemStatus<K, V, Ctx> targetStatus = this.getTargetStatus();
+                    for (int i = prevStatus.ordinal(); i < this.futures.length; i ++) {
+                        if (i > targetStatus.ordinal()) {
+                            this.futures[i].completeExceptionally(UNLOADED_EXCEPTION);
+                            this.futures[i] = UNLOADED_FUTURE;
+                        } else {
+                            this.futures[i] = this.futures[i].isDone() ? new CompletableFuture<>() : this.futures[i];
+                        }
+                    }
+                }
             } else if (compare > 0) { // status upgrade
                 Assertions.assertTrue(prevStatus.getNext() == status, "Invalid status upgrade");
 
@@ -292,19 +304,17 @@ public boolean setStatus(ItemStatus<K, V, Ctx> status, boolean isCancellation) {
         return true;
     }
 
-    void flushUnloadedStatus(ItemStatus<K, V, Ctx> startingPoint, boolean onDowngrade) {
+    void flushUnloadedStatus(ItemStatus<K, V, Ctx> currentStatus) {
         ArrayList<CompletableFuture<Void>> futuresToFire = null;
+        if (currentStatus.getNext() == null) {
+            return;
+        }
         synchronized (this.futures) {
-            final ItemStatus<K, V, Ctx> targetStatus = this.getTargetStatus();
-            for (int i = startingPoint.ordinal(); i < this.futures.length; i ++) {
-                if (i > targetStatus.ordinal()) {
-                    if (futuresToFire == null) futuresToFire = new ArrayList<>();
-                    CompletableFuture<Void> oldFuture = this.futures[i];
-                    futuresToFire.add(oldFuture);
-                    this.futures[i] = UNLOADED_FUTURE;
-                } else if (onDowngrade) {
-                    this.futures[i] = this.futures[i].isDone() ? new CompletableFuture<>() : this.futures[i];
-                }
+            for (int i = currentStatus.ordinal() + 1; i < this.futures.length; i ++) {
+                if (futuresToFire == null) futuresToFire = new ArrayList<>();
+                CompletableFuture<Void> oldFuture = this.futures[i];
+                futuresToFire.add(oldFuture);
+                this.futures[i] = UNLOADED_FUTURE;
             }
         }
         if (futuresToFire != null) {
@@ -315,6 +325,16 @@ void flushUnloadedStatus(ItemStatus<K, V, Ctx> startingPoint, boolean onDowngrad
         }
     }
 
+    void validateCompletedFutures(ItemStatus<K, V, Ctx> upTo) {
+        synchronized (this.futures) {
+            for (int i = this.unloadedStatus.ordinal() + 1; i <= upTo.ordinal(); i++) {
+                CompletableFuture<Void> future = this.futures[i];
+                Assertions.assertTrue(future != UNLOADED_FUTURE, "Future for loaded status cannot be UNLOADED_FUTURE");
+                Assertions.assertTrue(future.isDone(), "Future for status <= targetStatus must be completed");
+            }
+        }
+    }
+
     public synchronized void setDependencies(ItemStatus<K, V, Ctx> status, KeyStatusPair<K, V, Ctx>[] dependencies) {
         assertOpen();
         final int ordinal = status.ordinal();

src/main/java/com/ishland/flowsched/scheduler/StatusAdvancingScheduler.java

@@ -109,9 +109,11 @@ void tickHolder0(ItemHolder<K, V, Ctx, UserData> holder) {
             current = holder.getStatus();
             nextStatus = getNextStatus(current, holder.getTargetStatus());
             Assertions.assertTrue(holder.getStatus() == current);
+            holder.validateCompletedFutures(current);
 //        holder.sanitizeSetStatus = null;
             if (nextStatus == current) {
-                holder.flushUnloadedStatus(current, false);
+                holder.flushUnloadedStatus(current);
+                holder.validateCompletedFutures(current);
                 if (current.equals(getUnloadedStatus())) {
                     if (holder.isDependencyDirty()) {
                         holder.executeCriticalSectionAndBusy(() -> holder.cleanupDependencies(this));

src/test/java/com/ishland/flowsched/scheduler/SchedulerTest.java

@@ -1,13 +1,20 @@
 package com.ishland.flowsched.scheduler;
 
+import com.ishland.flowsched.scheduler.support.TestContext;
+import com.ishland.flowsched.scheduler.support.TestItem;
 import com.ishland.flowsched.scheduler.support.TestSchedulerImpl;
 import com.ishland.flowsched.scheduler.support.TestStatus;
+import org.junit.jupiter.api.Assertions;
 import org.junit.jupiter.api.Test;
 import org.junit.jupiter.api.extension.ExtendWith;
 import org.mockito.junit.jupiter.MockitoExtension;
 import org.mockito.junit.jupiter.MockitoSettings;
 import org.mockito.quality.Strictness;
 
+import java.util.ArrayList;
+import java.util.Random;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.atomic.AtomicBoolean;
 import java.util.concurrent.locks.LockSupport;
 
 @ExtendWith(MockitoExtension.class)
@@ -18,11 +25,27 @@ public class SchedulerTest {
     public void testSimple() {
         final TestSchedulerImpl scheduler = new TestSchedulerImpl();
         long startTime = System.nanoTime();
-//        scheduler.addTicket(100L, TestStatus.STATE_3, () -> {
-//            System.out.println("100 reached STATE_3 after " + (System.nanoTime() - startTime) + "ns");
-//            scheduler.removeTicket(100L, TestStatus.STATE_3);
-//        });
-        final long key = 1000L;
+
+        final long key = 1024L;
+        AtomicBoolean spamLoaderRunning = new AtomicBoolean(true);
+        ArrayList<CompletableFuture<Void>> spammedFutures = new ArrayList<>();
+
+        Thread spamLoader = new Thread(() -> {
+            Random random = new Random();
+            while (spamLoaderRunning.get()) {
+                long victim = random.nextLong(key - 1);
+                ItemHolder<Long, TestItem, TestContext, Void> holder = scheduler.addTicket(victim, TestStatus.STATE_8, StatusAdvancingScheduler.NO_OP);
+                CompletableFuture<Void> future = holder.getFutureForStatus0(TestStatus.STATE_8);
+                if (future.isCompletedExceptionally()) {
+                    Assertions.fail();
+                }
+                spammedFutures.add(future);
+                scheduler.removeTicket(victim, TestStatus.STATE_8);
+                LockSupport.parkNanos(1_000L);
+            }
+        });
+        spamLoader.start();
+
         scheduler.addTicket(key, TestStatus.STATE_7, () -> {
             System.out.println("reached STATE_7 after " + (System.nanoTime() - startTime) + "ns");
             scheduler.removeTicket(key, TestStatus.STATE_7);
@@ -33,6 +56,7 @@ public void testSimple() {
                 scheduler.addTicket(key, TestStatus.STATE_8, () -> {
                     System.out.println("reached STATE_8 after " + (System.nanoTime() - startTime) + "ns");
                     scheduler.removeTicket(key, TestStatus.STATE_8);
+                    spamLoaderRunning.set(false);
                 });
                 scheduler.getHolder(key).getFutureForStatus0(TestStatus.STATE_8).whenComplete((unused, throwable) -> {
                     if (throwable != null) throwable.printStackTrace();
@@ -62,6 +86,14 @@ public void testSimple() {
         }
 
         System.out.println("All unloaded after " + (System.nanoTime() - startTime) + "ns");
+
+        for (CompletableFuture<Void> spammedFuture : spammedFutures) {
+            if (!spammedFuture.isDone()) {
+                Assertions.fail();
+            }
+        }
+
+
 //        scheduler.shutdown();
     }