feat: add bounded cancellable payload work queue

build.gradle

@@ -81,6 +81,22 @@ if (providers.gradleProperty('interactionRegression').isPresent()) {
     sourceSets.main.java.srcDir('tests/integration')
     sourceSets.main.java.exclude('**/ChunkHandoffRegression.java')
 }
+if (providers.gradleProperty('ownershipRegression').isPresent()) {
+    if (providers.gradleProperty('interactionRegression').isPresent()) {
+        throw new GradleException('Run ownershipRegression and interactionRegression separately.')
+    }
+    sourceSets.main.java.srcDir('tests/ownership')
+    dependencies { runtimeOnly files(providers.gradleProperty('ccTestJar').get()) }
+    tasks.matching { it.name == 'runServer' }.configureEach {
+        doLast {
+            def output = file('runs/server/logs/latest.log').getText('UTF-8')
+            ['OWNERSHIP REGRESSION PASS', 'OWNERSHIP EDGE PASS', 'OWNERSHIP CALLBACK PASS', 'OWNERSHIP RESUME PASS'].each { marker ->
+                if (!output.contains(marker)) throw new GradleException("Missing runtime test result: ${marker}")
+            }
+            if (output.contains('Encountered an unexpected exception')) throw new GradleException('Runtime regression server crashed.')
+        }
+    }
+}
 
 tasks.withType(ProcessResources).configureEach {
     var replaceProperties = [
@@ -121,11 +137,12 @@ tasks.register('compileConcurrencyRegression', JavaCompile) {
             'src/main/java/dev/modzuozhi/core/compat/TaskBarrier.java',
             'src/main/java/dev/modzuozhi/core/compat/ManagedFutureAwait.java',
             'src/main/java/dev/modzuozhi/core/compat/PauseOwnership.java', 'tests/PauseOwnershipRegression.java',
-            'src/main/java/dev/modzuozhi/core/compat/TickProgress.java', 'tests/TickProgressRegression.java')
+            'src/main/java/dev/modzuozhi/core/compat/TickProgress.java', 'tests/TickProgressRegression.java',
+            'src/main/java/dev/modzuozhi/core/compat/PayloadWorkQueue.java', 'tests/PayloadWorkQueueRegression.java')
     classpath = files()
     destinationDirectory = regressionClasses
 }
-['TaskBarrierRegression', 'ManagedFutureAwaitRegression', 'PauseOwnershipRegression', 'TickProgressRegression'].each { regressionMain ->
+['TaskBarrierRegression', 'ManagedFutureAwaitRegression', 'PauseOwnershipRegression', 'TickProgressRegression', 'PayloadWorkQueueRegression'].each { regressionMain ->
     def regressionTask = tasks.register("run${regressionMain}", JavaExec) {
         dependsOn 'compileConcurrencyRegression'
         classpath = files(regressionClasses)

src/main/java/dev/modzuozhi/core/compat/PayloadWorkQueue.java

@@ -0,0 +1,92 @@
+package dev.modzuozhi.core.compat;
+
+import java.util.*;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.RejectedExecutionException;
+import java.util.function.LongSupplier;
+import java.util.function.Supplier;
+
+/** Bounded FIFO. Queue locks never encompass handlers or future completion callbacks. */
+public final class PayloadWorkQueue<K> {
+    private final LongSupplier clock;
+    private final LinkedHashSet<Entry<?>> pending = new LinkedHashSet<>();
+    private final Map<K, Integer> counts = new HashMap<>();
+    private Throwable closed;
+    private boolean draining;
+
+    public PayloadWorkQueue(LongSupplier nanoTime) { clock = Objects.requireNonNull(nanoTime); }
+
+    public <T> CompletableFuture<T> submit(K connection, Supplier<T> action) {
+        Objects.requireNonNull(connection);
+        Objects.requireNonNull(action);
+        Entry<T> entry = new Entry<>(connection, action);
+        Throwable rejected;
+        synchronized (this) {
+            rejected = closed;
+            if (rejected == null && (pending.size() >= 4096 || counts.getOrDefault(connection, 0) >= 256))
+                rejected = new RejectedExecutionException("Payload queue capacity exceeded");
+            if (rejected == null) {
+                pending.add(entry);
+                counts.merge(connection, 1, Integer::sum);
+            }
+        }
+        if (rejected != null) entry.result.completeExceptionally(rejected);
+        else entry.result.whenComplete((value, error) -> {
+            if (entry.result.isCancelled()) synchronized (this) { remove(entry); }
+        });
+        return entry.result;
+    }
+
+    private void remove(Entry<?> entry) {
+        if (pending.remove(entry)) counts.compute(entry.connection, (key, count) -> count == 1 ? null : count - 1);
+    }
+
+    /** Single consumer; recursive/concurrent drains are no-ops. Already running work is never interrupted. */
+    public int drainBatch() {
+        synchronized (this) {
+            if (draining || closed != null) return 0;
+            draining = true;
+        }
+        int ran = 0;
+        long start = clock.getAsLong();
+        try {
+            while (ran < 64 && clock.getAsLong() - start < 2_000_000L) {
+                Entry<?> entry;
+                synchronized (this) {
+                    if (closed != null || pending.isEmpty()) break;
+                    entry = pending.iterator().next();
+                    remove(entry);
+                }
+                entry.run();
+                ran++;
+            }
+            return ran;
+        } finally { synchronized (this) { draining = false; } }
+    }
+
+    public void close(Throwable reason) {
+        Objects.requireNonNull(reason);
+        List<Entry<?>> abandoned;
+        synchronized (this) {
+            if (closed != null) return;
+            closed = reason;
+            abandoned = new ArrayList<>(pending);
+            pending.clear(); counts.clear();
+        }
+        abandoned.forEach(entry -> entry.result.completeExceptionally(reason));
+    }
+
+    public synchronized int size() { return pending.size(); }
+
+    private final class Entry<T> {
+        final K connection;
+        final Supplier<T> action;
+        final CompletableFuture<T> result = new CompletableFuture<>();
+        Entry(K connection, Supplier<T> action) { this.connection = connection; this.action = action; }
+        void run() {
+            if (result.isCancelled()) return;
+            try { result.complete(action.get()); }
+            catch (Throwable error) { result.completeExceptionally(error); }
+        }
+    }
+}

tests/PayloadWorkQueueRegression.java

@@ -0,0 +1,81 @@
+import dev.modzuozhi.core.compat.PayloadWorkQueue;
+import java.util.*;
+import java.util.concurrent.*;
+import java.util.concurrent.atomic.*;
+
+public class PayloadWorkQueueRegression {
+    static void check(boolean ok, String why) { if (!ok) throw new AssertionError(why); }
+    public static void main(String[] args) throws Exception {
+        fifoAndExactlyOnce(); capacityAndCancellation(); batchCountAndTime();
+        closeAndReentrantCompletion(); concurrentProducers();
+        System.out.println("PAYLOAD QUEUE PASS");
+    }
+    static void fifoAndExactlyOnce() {
+        var q = new PayloadWorkQueue<String>(() -> 0L);
+        var seen = new ArrayList<Integer>();
+        var futures = new ArrayList<CompletableFuture<Integer>>();
+        for (int i=0;i<256;i++) { int n=i; futures.add(q.submit("a", () -> { seen.add(n); return n; })); }
+        while(q.size()>0) q.drainBatch();
+        check(seen.size()==256, "exactly once");
+        for(int i=0;i<256;i++) check(seen.get(i)==i && futures.get(i).join()==i, "FIFO/results");
+        check(q.drainBatch()==0, "no replay");
+    }
+    static void capacityAndCancellation() {
+        var q = new PayloadWorkQueue<Integer>(() -> 0L);
+        var fs = new ArrayList<CompletableFuture<Integer>>();
+        for(int c=0;c<16;c++) for(int i=0;i<256;i++) fs.add(q.submit(c, () -> 1));
+        check(q.size()==4096, "server capacity");
+        check(q.submit(99, () -> 2).isCompletedExceptionally(), "reject server overflow");
+        fs.getFirst().cancel(false);
+        check(q.size()==4095, "cancel returns capacity immediately");
+        check(!q.submit(0, () -> 3).isDone(), "cancel frees connection slot");
+        while(q.size()>0) q.drainBatch();
+        for(int i=0;i<256;i++) q.submit(7, () -> 1);
+        check(q.submit(7, () -> 9).isCompletedExceptionally(), "reject connection overflow");
+        check(!q.submit(8, () -> 1).isDone(), "other connection still allowed");
+        q.close(new IllegalStateException("stop"));
+    }
+    static void batchCountAndTime() {
+        var clock = new AtomicLong(); var q = new PayloadWorkQueue<String>(clock::get);
+        for(int i=0;i<65;i++) q.submit("a", () -> 1);
+        check(q.drainBatch()==64 && q.size()==1, "64 budget"); q.drainBatch();
+        var done = new AtomicInteger();
+        q.submit("a", () -> { clock.addAndGet(2_000_000); return done.incrementAndGet(); });
+        q.submit("a", done::incrementAndGet);
+        check(q.drainBatch()==1 && done.get()==1 && q.size()==1, "2ms boundary");
+        q.drainBatch();
+        var slow = q.submit("a", () -> { clock.addAndGet(100_000_000); return 42; });
+        check(q.drainBatch()==1 && slow.join()==42, "slow task not interrupted");
+    }
+    static void closeAndReentrantCompletion() {
+        var q = new PayloadWorkQueue<String>(() -> 0L);
+        var count = new AtomicInteger(); var f=q.submit("a", count::incrementAndGet);
+        f.thenRun(() -> q.submit("a", count::incrementAndGet)); q.drainBatch();
+        check(count.get()==2, "completion may enqueue");
+        var boom = new IllegalArgumentException("handler failure");
+        var bad=q.submit("a", () -> { throw boom; }); var good=q.submit("a", () -> 42);
+        q.drainBatch(); check(bad.isCompletedExceptionally() && good.join()==42, "isolate exceptions");
+        var pending=q.submit("a", count::incrementAndGet);
+        q.close(boom); check(pending.isCompletedExceptionally() && q.size()==0, "close settles pending");
+        check(q.submit("a", count::incrementAndGet).isCompletedExceptionally(), "close terminal");
+        q.drainBatch(); check(count.get()==2, "closed work never executes");
+        var q2=new PayloadWorkQueue<String>(() -> 0L);
+        var cancel=q2.submit("a", count::incrementAndGet); cancel.cancel(false); q2.drainBatch();
+        check(count.get()==2, "cancelled action skipped");
+    }
+    static void concurrentProducers() throws Exception {
+        var q = new PayloadWorkQueue<Integer>(() -> 0L);
+        var executed=new AtomicInteger(); var futures=new ConcurrentLinkedQueue<CompletableFuture<Integer>>();
+        var pool=Executors.newFixedThreadPool(8);
+        try {
+            var jobs=new ArrayList<Future<?>>();
+            for(int c=0;c<8;c++) { int key=c; jobs.add(pool.submit(() -> {
+                for(int i=0;i<250;i++) { var f=q.submit(key, executed::incrementAndGet); futures.add(f); if(i%2==0) f.cancel(false); }
+            })); }
+            for(var job:jobs) job.get(5,TimeUnit.SECONDS);
+            while(q.size()>0) q.drainBatch();
+            check(executed.get()==1000 && q.size()==0, "concurrent cancel/submit accounting");
+            check(futures.stream().allMatch(CompletableFuture::isDone), "no pending futures");
+        } finally { pool.shutdownNow(); }
+    }
+}