sfqd-core: fix refundable queue scaling and release 1.2.0

CHANGELOG.md

@@ -2,6 +2,31 @@
 
 All notable changes to this project are documented in this file.
 
+## 1.2.0 - 2026-09-30
+
+Removes per-flow queue scans from ordinary refundable admission and queued
+cancellation, preventing long queues from multiplying their lock-held work.
+
+### Changed
+
+- refund-mode enqueue and cancellation now update an AVL multiset of live
+  costs in `O(log K)` time, plus head-index maintenance where applicable;
+- exact queued tags are materialized on head promotion; dispatch adds
+  `O(log K)` per selected job to maintain the cost aggregate;
+- incremental sum/gcd admission preserves the exact existing refund-closure
+  domain, including denominator-factor removal after cancellation or dispatch;
+- JMH cancellation cycles now expose per-flow queue length as a parameter.
+
+### Verification and compatibility
+
+- adds long-queue differential, cost-aggregate, and cancellation-before-rebase
+  regression coverage;
+- preserves public signatures, exact scheduling order, numeric budgets, atomic
+  rejection, and default charge-reserved accounting;
+- canonical rebasing remains transactional and linear in queued jobs plus
+  registered flows; the scheduler still uses one lock;
+- no application-specific throughput or deadline guarantee is introduced.
+
 ## 1.1.0 - 2026-08-30
 
 Adds opt-in virtual-cost refunds for queued cancellation while preserving the

README.md

@@ -10,8 +10,8 @@ weights.
 
 ## Status
 
-Version 1.1.0 is the current stable release. The library is published to Maven
-Central as `io.github.pzhin:sfqd-core:1.1.0` and uses the Java package
+Version 1.2.0 is the current stable release. The library is published to Maven
+Central as `io.github.pzhin:sfqd-core:1.2.0` and uses the Java package
 `io.github.pzhin.sfqd`.
 
 The benchmark harness is an executable measurement protocol, not a benchmark
@@ -57,7 +57,7 @@ Then add it to a local consumer:
 <dependency>
   <groupId>io.github.pzhin</groupId>
   <artifactId>sfqd-core</artifactId>
-  <version>1.1.0</version>
+  <version>1.2.0</version>
 </dependency>
 ```
 
@@ -248,7 +248,8 @@ job's cost from future virtual debt. Every later queued job of the same flow is
 recomputed in enqueue order from the cancelled job's start tag, and the flow's
 finish history moves to the end of that recomputed suffix. Other flows do not
 change. The recomputation is one atomic transition under the scheduler's common
-lock; callers cannot observe a partial suffix.
+lock; callers cannot observe a partial suffix. The implementation represents
+this transition lazily and materializes tags when a job becomes the flow head.
 
 Refund admission is intentionally narrower. Before accepting a job, enqueue
 checks that the prospective flow queue has one common exact denominator and a
@@ -415,26 +416,34 @@ recorded run is reviewed for the target hardware and workload.
 ## Complexity
 
 Let `R` be registered flows, `Q` queued jobs, `B` backlogged flows, `K` jobs in
-the affected per-flow queue or suffix, and `m` the number of jobs returned by
+the affected per-flow queue, and `m` the number of jobs returned by
 one capacity call.
 
 | Operation | Expected or worst-case time |
 | --- | ---: |
 | register or close flow | expected `O(1)` |
 | charge-reserved enqueue to a backlogged flow | expected `O(1)` |
 | charge-reserved enqueue that makes a flow backlogged | `O(log B)` |
-| refund enqueue to a backlogged flow | `O(K)` |
-| refund enqueue that makes a flow backlogged | `O(K + log B)` |
+| refund enqueue to a backlogged flow | `O(log K)` |
+| refund enqueue that makes a flow backlogged | `O(log K + log B)` |
 | charge-reserved cancel of a non-head queued job | expected `O(1)` |
 | charge-reserved cancel of a flow head | `O(log B)` |
-| refund cancel without changing the indexed head | `O(K)` |
-| refund cancel that changes the indexed head | `O(K + log B)` |
-| dispatch `m` jobs | `O(m log B + m)` |
+| refund cancel without changing the indexed head | `O(log K)` |
+| refund cancel that changes the indexed head | `O(log K + log B)` |
+| charge-reserved dispatch `m` jobs | `O(m log B + m)` |
+| refund dispatch `m` jobs | `O(m (log B + log K) + m)` |
 | ordinary completion | expected `O(1)` |
 | aggregate or per-flow snapshot | expected `O(1)` |
 | transition to global idle | `O(R)` |
 | rare exact-tag normalization | `O(Q + R)` time and temporary space |
 
+Refund accounting materializes tags only when a job becomes its flow's head.
+An AVL multiset of live queued costs maintains their exact sum and gcd, so
+ordinary admission and cancellation never scan the queue. Numeric admission
+and scheduling semantics are unchanged; rare canonical rebases still visit
+all queued jobs. Both snapshot methods take the scheduler lock but do not
+scan jobs.
+
 Retained state is `O(Q + running jobs + R)`. Terminal jobs and payloads are
 not retained as tombstones. These are algorithmic complexity bounds, not
 measured performance claims; no throughput or latency claim is made for refund

docs/FORMAL_SPEC.md

@@ -687,14 +687,30 @@ cancellation. At the successful cancel LP:
 6. if the indexed flow head changed, insert the new head into the global
    priority index only after its recomputed tag is installed.
 
-The complete removal and suffix recomputation are one atomic transition under
+The complete removal and logical suffix recomputation are one atomic transition under
 the scheduler's serialization boundary. No partially recomputed suffix is
 observable. The refund-closure invariant established at enqueue makes every
 exact cancellation computation fit the existing budgets, so `cancel` gains no
 numeric rejection or exception. Other flows and their tags do not change,
 intra-flow enqueue order is preserved, every queued `S` remains at least `V`,
 and no past `dispatchUpTo` result is reconsidered.
 
+An implementation may represent this exact transition lazily. While a flow is
+backlogged, every successor starts at its predecessor's finish: `V` cannot
+pass the indexed head. Therefore cancellation subtracts `cost(c)/weight`
+from `lastFinish`, preserves the head start (including head cancellation),
+and removes the cost from the live chain. Head promotion materializes the
+next exact tag without consulting the then-current `V`. This is a representation
+optimization, not a change to the above transition or admission domain.
+
+For incremental admission, with total queued cost `C` and gcd `G` of the
+individual costs, the increment-denominator lcm is `weight/gcd(weight,G)`.
+Combining this with the head denominator yields exactly the `L` in §3.2.2;
+the final numerator is `baseStart * L + C * L / weight` using exact rational
+scaling. The gcd must include individual costs, not only their sum. Deletion
+must remove obsolete denominator factors. A rebase still validates every
+logical queued tag and the transformed closure before any state is committed.
+
 ### 7.5 `dispatchUpTo(k)` / `dispatch(k)`
 
 `k` is an integer in `0..D`; a negative value or `k>D` is an invalid argument.

docs/THEORY.md

@@ -96,9 +96,10 @@ work while preserving the ordering within each busy period.
 Cancellation is a library extension rather than a result supplied by the SFQ
 or SFQ(D) papers. The default policy keeps the tags assigned at admission, so a
 cancelled queued cost remains virtual debt until global idle. The opt-in refund
-policy instead recomputes the later queued suffix of that flow at the
+policy instead logically recomputes the later queued suffix of that flow at the
 cancellation linearization point. It does not undo an earlier dispatch or
-rewrite virtual-time history.
+rewrite virtual-time history. The implementation may defer tag materialization
+until head promotion while preserving this exact transition.
 
 Exact refund must not turn cancellation into a fallible cleanup operation. The
 refund policy therefore narrows admission: before accepting a job, the library

docs/TOOLING.md

@@ -153,7 +153,7 @@ Before running the profile, the release engineer must have:
 - an unlocked GnuPG agent, or `MAVEN_GPG_PASSPHRASE` supplied through a secure
   environment mechanism.
 
-Publish the immutable 1.1.0 coordinates from the commit tagged `v1.1.0`:
+Publish the immutable 1.2.0 coordinates from the commit tagged `v1.2.0`:
 
 ```shell
 ./mvnw --batch-mode --no-transfer-progress -Prelease clean deploy

pom.xml

@@ -6,7 +6,7 @@
   <modelVersion>4.0.0</modelVersion>
   <groupId>io.github.pzhin</groupId>
   <artifactId>sfqd-java-parent</artifactId>
-  <version>1.1.0</version>
+  <version>1.2.0</version>
   <packaging>pom</packaging>
   <name>SFQ(D) Java Parent</name>
   <description>Build parent for the SFQ(D) Java library.</description>
@@ -35,7 +35,7 @@
     <connection>scm:git:https://github.com/pzhin/sfqd-java.git</connection>
     <developerConnection>scm:git:ssh://git@github.com/pzhin/sfqd-java.git</developerConnection>
     <url>https://github.com/pzhin/sfqd-java</url>
-    <tag>v1.1.0</tag>
+    <tag>v1.2.0</tag>
   </scm>
 
   <modules>
@@ -289,7 +289,7 @@
               </excludeArtifacts>
               <autoPublish>true</autoPublish>
               <waitUntil>published</waitUntil>
-              <deploymentName>sfqd-java 1.1.0</deploymentName>
+              <deploymentName>sfqd-java 1.2.0</deploymentName>
             </configuration>
           </plugin>
         </plugins>

sfqd-benchmarks/README.md

@@ -75,7 +75,11 @@ metric and the normalization.
 ### Cancellation cycles
 
 `CancellationCycleBenchmark` measures `cancel -> enqueue replacement`. The
-score belongs to the combined cycle, not cancellation alone. The
+score belongs to the combined cycle, not cancellation alone. `queueLength`
+selects 2, 250, 2,500, or 20,000 jobs per active flow. For a bounded long-queue
+comparison use `-p flowCount=1 -p depth=16 -p scenario=UNIFORM -p target=HEAD`
+and vary `queueLength` under both policies; do not combine every large flow
+count with every large per-flow queue on a memory-limited machine. The
 `cancellationAccounting` parameter selects `CHARGE_RESERVED_COST` or
 `REFUND_CANCELLED_COST`; the same policy dimension is present in isolated
 head/non-head cancellation latency, terminal idle-reset cancellation, and

sfqd-benchmarks/pom.xml

@@ -6,7 +6,7 @@
   <parent>
     <groupId>io.github.pzhin</groupId>
     <artifactId>sfqd-java-parent</artifactId>
-    <version>1.1.0</version>
+    <version>1.2.0</version>
   </parent>
   <artifactId>sfqd-benchmarks</artifactId>
   <name>SFQ(D) JMH Harness</name>

sfqd-benchmarks/src/main/java/io/github/pzhin/sfqd/benchmarks/CancellationCycleBenchmark.java

@@ -47,6 +47,10 @@ public static class CancellationState {
         @Param({"1", "8", "64", "256"})
         private int depth;
 
+        /** Queued jobs per active flow. */
+        @Param({"2", "250", "2500", "20000"})
+        private int queueLength = 2;
+
         /** Deterministic workload distribution. */
         @Param
         private Scenario scenario;
@@ -72,7 +76,7 @@ public static class CancellationState {
         /** Creates the bounded caller and scheduler queues outside measurements. */
         @Setup(Level.Trial)
         public void setupTrial() {
-            fixture = new Fixture(flowCount, depth, scenario, 0, 2, cancellationAccounting);
+            fixture = new Fixture(flowCount, depth, scenario, 0, queueLength, cancellationAccounting);
         }
 
         /** Selects and removes caller-side target bookkeeping outside the timed cycle. */

sfqd-core/pom.xml

@@ -6,7 +6,7 @@
   <parent>
     <groupId>io.github.pzhin</groupId>
     <artifactId>sfqd-java-parent</artifactId>
-    <version>1.1.0</version>
+    <version>1.2.0</version>
   </parent>
   <artifactId>sfqd-core</artifactId>
   <name>SFQ(D) Core</name>

sfqd-core/src/main/java/io/github/pzhin/sfqd/RefundCosts.java

@@ -0,0 +1,143 @@
+package io.github.pzhin.sfqd;
+
+import java.math.BigInteger;
+import java.util.Objects;
+
+/** AVL multiset of queued costs; retains only live costs, their sum, and their gcd. */
+final class RefundCosts {
+    private Node root;
+    private BigInteger sum = BigInteger.ZERO;
+
+    BigInteger sum() {
+        return sum;
+    }
+
+    BigInteger gcd() {
+        return BigInteger.valueOf(gcd(root));
+    }
+
+    void add(long cost) {
+        root = add(root, cost);
+        sum = sum.add(BigInteger.valueOf(cost));
+    }
+
+    void remove(long cost) {
+        root = remove(root, cost, false);
+        sum = sum.subtract(BigInteger.valueOf(cost));
+    }
+
+    private static Node add(Node node, long cost) {
+        if (node == null) {
+            return new Node(cost);
+        }
+        if (cost < node.cost) {
+            node.left = add(node.left, cost);
+        } else if (cost > node.cost) {
+            node.right = add(node.right, cost);
+        } else {
+            node.count++;
+        }
+        return balance(node);
+    }
+
+    private static Node remove(Node node, long cost, boolean all) {
+        if (node == null) {
+            throw new IllegalStateException("queued cost missing");
+        }
+        if (cost < node.cost) {
+            node.left = remove(node.left, cost, all);
+        } else if (cost > node.cost) {
+            node.right = remove(node.right, cost, all);
+        } else if (!all && node.count > 1) {
+            node.count--;
+        } else {
+            if (node.left == null) {
+                return node.right;
+            }
+            if (node.right == null) {
+                return node.left;
+            }
+            Node successor = node.right;
+            while (successor.left != null) {
+                successor = successor.left;
+            }
+            node.cost = successor.cost;
+            node.count = successor.count;
+            node.right = remove(node.right, successor.cost, true);
+        }
+        return balance(node);
+    }
+
+    private static Node balance(Node node) {
+        update(node);
+        if (height(node.left) - height(node.right) > 1) {
+            Node left = Objects.requireNonNull(node.left);
+            if (height(left.left) < height(left.right)) {
+                node.left = rotateLeft(node.left);
+            }
+            return rotateRight(node);
+        }
+        if (height(node.right) - height(node.left) > 1) {
+            Node right = Objects.requireNonNull(node.right);
+            if (height(right.right) < height(right.left)) {
+                node.right = rotateRight(node.right);
+            }
+            return rotateLeft(node);
+        }
+        return node;
+    }
+
+    private static Node rotateLeft(Node node) {
+        Node promoted = Objects.requireNonNull(node.right);
+        node.right = promoted.left;
+        promoted.left = node;
+        update(node);
+        update(promoted);
+        return promoted;
+    }
+
+    private static Node rotateRight(Node node) {
+        Node promoted = Objects.requireNonNull(node.left);
+        node.left = promoted.right;
+        promoted.right = node;
+        update(node);
+        update(promoted);
+        return promoted;
+    }
+
+    private static void update(Node node) {
+        node.height = 1 + Math.max(height(node.left), height(node.right));
+        node.gcd = gcd(node.cost, gcd(gcd(node.left), gcd(node.right)));
+    }
+
+    private static int height(Node node) {
+        return node == null ? 0 : node.height;
+    }
+
+    private static long gcd(Node node) {
+        return node == null ? 0L : node.gcd;
+    }
+
+    private static long gcd(long first, long second) {
+        while (second != 0L) {
+            long remainder = first % second;
+            first = second;
+            second = remainder;
+        }
+        return first;
+    }
+
+    private static final class Node {
+        private long cost;
+        private int count = 1;
+        private int height = 1;
+        private long gcd;
+        private Node left;
+        private Node right;
+
+        private Node(long cost) {
+            this.cost = cost;
+            gcd = cost;
+        }
+    }
+}