Transient Work with CompletableFuture and crash recovery (PR #118)

Work is now a plain POJO — no longer an @entity or PanacheEntity.
Work items exist only in memory, eliminating orphaned work rows and
removing the work/work_values/work_active_nodes/work_source_nodes
tables entirely.

CompletableFuture upload tracking:
- upload() returns CompletableFuture<Void> that completes when all
  work (including cascaded children) finishes
- UploadTracker with AtomicInteger tracks pending work count per upload
- Tracker association derived from Work.sourceValues (no duplicated
  state), supporting future cross-upload Work automatically
- createTracked() accepts Set<Long> rootValueIds for multi-source
- Double-decrement protection on retry path
- UploadTracker.fail() poisons counter to prevent subsequent complete()

Crash recovery:
- UploadProcessingEntity tracks each upload (rootValueId, folderName)
- On startup, incomplete uploads are re-triggered using all source
  nodes (not just top-level) to handle mid-cascade crashes
- @priority ordering guarantees WorkService initializes before recovery
- Value deduplication in execute() safely skips already-computed values

Source value data kept in memory through work queue:
- Avoids DB round-trip JSON deserialization (was #1 CPU hotspot)
- Hibernate.isPropertyInitialized() for conditional lazy loading fallback

FolderServiceTest (5 tests):
- Recovery reprocesses incomplete upload
- Recovery skips missing root value / missing folder
- Recovery processes multiple independent nodes
- Recovery completes when all values already computed

Also: LoadLegacyRuns uses CompletableFuture.allOf() for batched uploads,
FolderResource upload description fixed, unused code removed.

Author: stalep

src/main/java/io/hyperfoil/tools/h5m/api/svc/FolderServiceInterface.java

@@ -9,6 +9,7 @@
 import java.nio.file.Path;
 import java.util.List;
 import java.util.Map;
+import java.util.concurrent.CompletableFuture;
 
 /**
  * Service interface for managing Folders.
@@ -55,12 +56,15 @@ public interface FolderServiceInterface {
 
     /**
      * Uploads data to a specific path within a folder.
+     * Returns a CompletableFuture that completes when all processing
+     * (including cascaded work) finishes for this upload.
      *
      * @param name The name of the folder.
      * @param path The path within the folder.
      * @param data The JSON data to upload.
+     * @return A future that completes when all work for this upload is done.
      */
-    void upload(String name, String path, JsonNode data);
+    CompletableFuture<Void> upload(String name, String path, JsonNode data);
 
     /**
      * Recalculates the contents or state of a folder by its name.

src/main/java/io/hyperfoil/tools/h5m/cli/LoadLegacyRuns.java

@@ -15,9 +15,13 @@
 import java.sql.PreparedStatement;
 import java.sql.ResultSet;
 import java.sql.Statement;
+import java.util.ArrayList;
 import java.util.HashMap;
+import java.util.List;
 import java.util.Map;
 import java.util.Scanner;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.TimeUnit;
 import java.util.concurrent.Callable;
 
 @CommandLine.Command(name="load-legacy-runs")
@@ -112,27 +116,36 @@ public Integer call() throws Exception {
                     int count = 0;
                     int batchCount = 0;
                     Scanner scanner = new Scanner(System.in);
+                    List<CompletableFuture<Void>> batchFutures = new ArrayList<>();
                     try (ResultSet rs = ps.executeQuery()) {
                         while(rs.next()){
                             Long id = rs.getLong(1);
                             System.out.println(name+" "+id);
                             JsonNode data = mapper.readTree(rs.getCharacterStream("data"));
-                            folderService.upload(folder.name(),null,data);
+                            batchFutures.add(folderService.upload(folder.name(),null,data));
                             count++;
                             batchCount++;
-                            if(batch > 0 && batchCount > batch){
-                                System.out.println("waiting for batch to complete");
-                                while(!workService.isIdle()){
-                                    Thread.sleep(10_000);
-                                }
+                            if(batch > 0 && batchCount >= batch){
+                                System.out.println("waiting for batch of " + batchCount + " to complete");
+                                CompletableFuture.allOf(batchFutures.toArray(new CompletableFuture[0]))
+                                        .orTimeout(10, TimeUnit.MINUTES)
+                                        .join();
                                 System.out.println("batch complete");
+                                batchFutures.clear();
                                 if(pause){
                                     scanner.nextLine();
                                 }
                                 batchCount = 0;
                             }
                         }
                     }
+                    // Wait for any remaining uploads
+                    if (!batchFutures.isEmpty()) {
+                        System.out.println("waiting for final " + batchFutures.size() + " uploads to complete");
+                        CompletableFuture.allOf(batchFutures.toArray(new CompletableFuture[0]))
+                                .orTimeout(10, TimeUnit.MINUTES)
+                                .join();
+                    }
                     System.out.println("loaded " + count + " runs");
                 } finally {
                     connection.setAutoCommit(true);

src/main/java/io/hyperfoil/tools/h5m/entity/UploadProcessingEntity.java

@@ -0,0 +1,35 @@
+package io.hyperfoil.tools.h5m.entity;
+
+import io.quarkus.hibernate.orm.panache.PanacheEntity;
+import jakarta.persistence.*;
+import org.hibernate.annotations.CreationTimestamp;
+
+import java.time.LocalDateTime;
+
+/**
+ * Tracks whether all work for a given upload has been processed.
+ * Used for crash recovery: on startup, incomplete uploads are re-triggered.
+ */
+@Entity(name = "upload_processing")
+public class UploadProcessingEntity extends PanacheEntity {
+
+    @Column(name = "root_value_id", nullable = false, updatable = false)
+    public long rootValueId;
+
+    @Column(name = "folder_name", nullable = false, updatable = false)
+    public String folderName;
+
+    @Column(nullable = false)
+    public boolean completed = false;
+
+    @CreationTimestamp
+    @Column(updatable = false)
+    public LocalDateTime createdAt;
+
+    public UploadProcessingEntity() {}
+
+    public UploadProcessingEntity(long rootValueId, String folderName) {
+        this.rootValueId = rootValueId;
+        this.folderName = folderName;
+    }
+}

src/main/java/io/hyperfoil/tools/h5m/entity/work/Work.java

@@ -4,53 +4,25 @@
 import io.hyperfoil.tools.h5m.entity.ValueEntity;
 import io.hyperfoil.tools.h5m.entity.node.RelativeDifference;
 import io.hyperfoil.tools.h5m.svc.WorkService;
-import io.quarkus.hibernate.orm.panache.PanacheEntity;
 import jakarta.enterprise.inject.spi.CDI;
-import jakarta.persistence.*;
-import org.hibernate.annotations.BatchSize;
 
 import java.util.*;
 import java.util.stream.Collectors;
 
 
-@Entity(name = "work")
-@DiscriminatorColumn(name = "type")
-@DiscriminatorValue("node")
 //cross test comparison could use sourceNodes and not have an activeNode?
 //custom post nodegroup actions could have sourceNodes without activeNode
-public class Work  extends PanacheEntity implements Runnable, Comparable<Work>{
-
-    @BatchSize(size=10)
-    @ManyToMany(cascade = {CascadeType.PERSIST}, fetch = FetchType.LAZY)
-    @JoinTable(
-            name="work_values",
-            joinColumns = @JoinColumn(name = "work_id"),
-            inverseJoinColumns = @JoinColumn(name = "value_id")
-    )
+public class Work implements Runnable, Comparable<Work>{
+
     public List<ValueEntity> sourceValues;//multiple values could happen for cross test comparisons and
 
-    @BatchSize(size=10)
-    @ManyToMany(cascade = {CascadeType.PERSIST}, fetch = FetchType.LAZY)
-    @JoinTable(
-            name="work_source_nodes",
-            joinColumns = @JoinColumn(name = "work_id"),
-            inverseJoinColumns = @JoinColumn(name = "node_id")
-    )
     public List<NodeEntity> sourceNodes; //what is going to use a list of sources that are not already listed for the activeNode?
 
     public int retryCount;
 
-    @BatchSize(size=10)
-    @ManyToMany(cascade = {CascadeType.PERSIST}, fetch = FetchType.EAGER)
-    @JoinTable(
-            name="work_active_nodes",
-            joinColumns = @JoinColumn(name = "work_id"),
-            inverseJoinColumns = @JoinColumn(name = "node_id")
-    )
     public Set<NodeEntity> activeNodes;
 
-    boolean cumulative = false;
-
+    public boolean cumulative = false;
 
     public Work(){
         retryCount = 0;
@@ -187,7 +159,7 @@ public int compareTo(Work o) {
 
     @Override
     public String toString() {
-        return "Work<id="+id+" activeNodes="+activeNodes+
+        return "Work<activeNodes="+activeNodes+
                 " sourceNodes="+sourceNodes.stream().map(n->""+n.getId()).collect(Collectors.joining(","))+
                 " sourceValues="+sourceValues.stream().map(v->""+v.getId()).collect(Collectors.joining(","))+
                 " retry="+retryCount+

src/main/java/io/hyperfoil/tools/h5m/queue/UploadTracker.java

@@ -0,0 +1,54 @@
+package io.hyperfoil.tools.h5m.queue;
+
+import io.quarkus.logging.Log;
+
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.atomic.AtomicInteger;
+
+/**
+ * Tracks completion of all work items triggered by a single upload.
+ * The counter is incremented when work items are created (including cascade)
+ * and decremented when each work item finishes. The future completes
+ * when all work reaches zero, or completes exceptionally on failure.
+ */
+public class UploadTracker {
+
+    private final long rootValueId;
+    private final AtomicInteger pendingCount = new AtomicInteger(0);
+    private final CompletableFuture<Void> future = new CompletableFuture<>();
+
+    public UploadTracker(long rootValueId) {
+        this.rootValueId = rootValueId;
+    }
+
+    public CompletableFuture<Void> getFuture() {
+        return future;
+    }
+
+    public void increment(int count) {
+        pendingCount.addAndGet(count);
+    }
+
+    public void decrement() {
+        int remaining = pendingCount.decrementAndGet();
+        Log.debugf("UploadTracker[%d]: decrement → %d remaining", rootValueId, remaining);
+        if (remaining == 0) {
+            future.complete(null);
+        } else if (remaining < 0) {
+            Log.warnf("UploadTracker[%d]: over-decremented to %d — possible accounting bug", rootValueId, remaining);
+        }
+    }
+
+    public void fail(Throwable t) {
+        Log.errorf(t, "UploadTracker[%d]: work failed", rootValueId);
+        // Set pending to a negative sentinel so subsequent decrements
+        // cannot trigger future.complete(null) — the failure wins
+        pendingCount.set(Integer.MIN_VALUE);
+        future.completeExceptionally(t);
+    }
+
+    @Override
+    public String toString() {
+        return "UploadTracker[rootValueId=" + rootValueId + ", pending=" + pendingCount.get() + "]";
+    }
+}

src/main/java/io/hyperfoil/tools/h5m/queue/WorkQueue.java

@@ -188,8 +188,8 @@ public Collection<Work> addWorks(Collection<Work> works){
             List<Work> acceptedWork = works.stream().filter(w -> {
                 boolean has = hasWork(w);
                 if (has) {
-                    log.warn("addWorks: REJECTED duplicate work id={} hash={} pending={} active={}",
-                            w.id, w.hashCode(), isPending(w), isActive(w));
+                    log.warn("addWorks: REJECTED duplicate work hash={} pending={} active={}",
+                            w.hashCode(), isPending(w), isActive(w));
                 }
                 return !has;
             }).peek(w-> {

src/main/java/io/hyperfoil/tools/h5m/rest/FolderResource.java

@@ -83,6 +83,9 @@ public void upload(
             @QueryParam("path") @Parameter(description = "Path within the folder") String path,
             JsonNode data) {
         folderService.upload(name, path, data);
+        // The upload() now returns a CompletableFuture, but the REST endpoint
+        // doesn't need to wait for it — the caller can poll for results.
+        // Future enhancement: return CompletionStage<Void> for async response.
     }
 
     @POST