core: add denominator-safe weight domains

README.md

@@ -95,15 +95,22 @@ deterministic FIFO tie-break. No floating-point arithmetic is used.
 ## Minimal example
 
 ```java
+import io.github.pzhin.sfqd.CancellationAccounting;
 import io.github.pzhin.sfqd.CompletionResult;
 import io.github.pzhin.sfqd.Dispatch;
 import io.github.pzhin.sfqd.EnqueueResult;
 import io.github.pzhin.sfqd.RegisterFlowResult;
 import io.github.pzhin.sfqd.SchedulerConfig;
 import io.github.pzhin.sfqd.SfqdScheduler;
+import io.github.pzhin.sfqd.WeightDomain;
 
 var scheduler = new SfqdScheduler<String, String, Runnable>(
-        new SchedulerConfig(4, 1_000, 100_000));
+        new SchedulerConfig(
+                4,
+                1_000,
+                100_000,
+                CancellationAccounting.CHARGE_RESERVED_COST,
+                WeightDomain.divisorsOf(8)));
 
 var result = scheduler.registerFlow("tenant-a", 2);
 if (!(result instanceof RegisterFlowResult.Registered registered)) {
@@ -164,9 +171,9 @@ whose later jobs could be stranded if an earlier pool submission throws.
 ### Register a flow
 
 `registerFlow(flowId, weight)` creates an opaque flow handle. A flow identifier
-cannot be registered twice at the same time, and the configured flow limit is
-enforced. A closed identifier may later be registered again, producing a new
-handle.
+cannot be registered twice at the same time, and the configured flow and weight
+domain limits are enforced. A closed identifier may later be registered again,
+producing a new handle.
 
 ### Enqueue a job
 
@@ -326,6 +333,34 @@ new SchedulerConfig(
 
 No free-cancellation accounting policy is currently implemented.
 
+The three- and four-argument forms preserve the unrestricted positive `long`
+weight domain. For production configurations with a known common scale, the
+five-argument form can reject weights outside a denominator-safe profile at
+registration:
+
+```java
+new SchedulerConfig(
+        issueDepth,
+        maxFlows,
+        maxLiveJobs,
+        CancellationAccounting.CHARGE_RESERVED_COST,
+        WeightDomain.divisorsOf(8));
+```
+
+This profile accepts `8, 4, 2, 1, 1`: every weight divides `8`. Consequently,
+every reduced `cost / weight` denominator divides `8`, and exact addition,
+maximum, and rebase subtraction cannot introduce new denominator factors.
+`WeightDomain.unrestricted()` remains available for workloads that need the
+full `long` range.
+
+The divisor profile prevents denominator growth caused by mutually coprime
+weights; it is not an unconditional promise that `NUMERIC_LIMIT` can never
+occur. Numerators and accumulated cancellation debt still use the documented
+finite exact-arithmetic budget. In unrestricted mode, a natural trace with an
+anchor job and 69 consecutive prime weights above `2^60` reaches
+`NUMERIC_LIMIT` after 273 successful admissions; this behavior is covered by a
+regression test.
+
 Weights and costs are positive `long` values. Choose `D` as the number of jobs
 your execution layer can have issued but not completed. For `N` identical
 non-preemptive resources, the usual direct mapping is `D = N`.

docs/FORMAL_SPEC.md

@@ -47,6 +47,11 @@ The operation is not a notification that capacity has become available.
 - `cost` — the supplied job cost, an integer in `1..Long.MAX_VALUE`.
 - `weight` — the flow weight, an integer in `1..Long.MAX_VALUE`.
 
+The configured `weightDomain` MAY retain that unrestricted syntactic domain or
+restrict successful registration to positive divisors of one fixed common
+scale `W` in `1..Long.MAX_VALUE`. A weight outside the configured domain is an
+operational registration rejection, not an invalid Java argument.
+
 `FlowId` and `JobId` MUST honor the Java `equals/hashCode` contract throughout
 the stated lifetime. Both methods MUST be deterministic, side-effect-free, and
 non-reentrant with respect to the scheduler; they MUST NOT invoke scheduler
@@ -69,8 +74,10 @@ Fairness is defined in terms of `cost`, not unknown actual execution time.
   `D..Integer.MAX_VALUE`.
 - `cancellationAccounting` — the fixed policy
   `CancellationAccounting.CHARGE_RESERVED_COST`; no alternative policy exists.
+- `weightDomain` — either unrestricted positive `long` weights or the positive
+  divisors of one fixed common scale `W`.
 
-All four values are immutable after instance construction. Null values,
+All five values are immutable after instance construction. Null values,
 out-of-range values, and `maxLiveJobs < D` are rejected before an observable
 instance exists.
 
@@ -172,9 +179,9 @@ transaction. Silent overflow, rounding, partial rebase, and order changes are
 forbidden.
 
 When `V=0` and `lastFinish=0`, a registered flow MUST accept one job with any
-`cost,weight` pair in `1..Long.MAX_VALUE`, including the maximum values, unless
-an independent identity or live-item limit applies: a single reduced fraction
-occupies at most 63 bits in either component.
+`cost` in `1..Long.MAX_VALUE` and its already accepted registration weight,
+unless an independent identity or live-item limit applies: a single reduced
+fraction occupies at most 63 bits in either component.
 
 The number of successfully accepted jobs over an instance lifetime is limited
 to `Long.MAX_VALUE`. State stores `lastJobSequence` in
@@ -206,6 +213,23 @@ The number of read-only and failed calls is not limited by scheduler state. The
 number of successful cancel, dispatch, and complete calls is limited by the
 number of accepted incarnations.
 
+### 3.2.1 Divisor-constrained weight domain
+
+For `weightDomain = divisorsOf(W)`, registration accepts a weight `w` only when
+`W mod w = 0`. For every accepted job, the reduced denominator of
+`cost / weight` therefore divides `W`. The set of rational numbers whose
+reduced denominators divide `W` is closed under exact addition, subtraction,
+and maximum. It follows inductively that every stored `S`, `F`, `V`,
+`lastFinish`, and every exact rebase result has a reduced denominator dividing
+`W`; distinct registered weights cannot accumulate new prime denominator
+factors.
+
+This is a denominator-safety property, not an infinite-lifetime numeric
+guarantee. Numerators remain subject to the 4096-bit persistent budget, and the
+charge-reserved cancellation policy can accumulate finish-tag debt during a
+continuous busy period. `NUMERIC_LIMIT` therefore remains a permitted enqueue
+result under a divisor-constrained domain.
+
 ### 3.3 Exact rebasing
 
 A rebase is a representational substitution of the same semantic state. Let
@@ -504,16 +528,17 @@ consistently.
 Order of processing:
 
 1. Validate a non-null `flowId` and `weight` in `1..Long.MAX_VALUE`.
-2. If `flowId` is in `RegisteredById`, return
+2. If `weight` is outside `weightDomain`, return `WEIGHT_OUTSIDE_DOMAIN`.
+3. If `flowId` is in `RegisteredById`, return
    `DUPLICATE_REGISTERED_ID`.
-3. If the registered count equals `maxFlows`, return `FLOW_LIMIT`.
-4. If `lastFlowSequence == Long.MAX_VALUE`, return
+4. If the registered count equals `maxFlows`, return `FLOW_LIMIT`.
+5. If `lastFlowSequence == Long.MAX_VALUE`, return
    `FLOW_SEQUENCE_EXHAUSTED`.
-5. Create an inert `FlowHandle(ownerToken,lastFlowSequence+1)` and FlowState
+6. Create an inert `FlowHandle(ownerToken,lastFlowSequence+1)` and FlowState
    with `lastFinish=0`, zero counts, `acceptedCost=0`, `dispatchedCost=0`,
    `cancelledCost=0`, `runningSuppliedCost=0`, and the fixed weight; insert
    both registration indexes and update the sequence.
-6. Return `REGISTERED(flowHandle)`.
+7. Return `REGISTERED(flowHandle)`.
 
 Registration during a non-empty busy period is allowed: it does not affect
 scheduling before the first enqueue. A rejection leaves state unchanged and
@@ -937,7 +962,7 @@ not transitively retain the scheduler.
 | Batch dispatch | Filling depth is described without API atomicity | One call selects a sequential SFQ(D) batch but linearizes as a whole |
 | Completion order | Black-box server | Any running handle may complete; the scheduler imposes no completion order |
 | Weight changes | Undefined | Weight is fixed for the registration lifetime; change requires safe close and a new registration |
-| Numbers | Mathematical unbounded tags | Exact canonical rationals, fail-closed 4096-bit persistent and 8193-bit transient budgets, and transactional all-registration rebase |
+| Numbers | Mathematical unbounded tags | Exact canonical rationals, fail-closed 4096-bit persistent and 8193-bit transient budgets, transactional all-registration rebase, and an optional common-scale divisor weight domain that prevents new denominator factors but does not bound numerators or cancellation debt |
 | Retention | Not considered | Payload release, no terminal metadata, and `maxLiveJobs` and `maxFlows` bounds |
 | Introspection | Not considered | Exact atomic aggregate and per-registration lifecycle snapshots without tags, identifiers, or a clock |
 | Executor rejection | Not considered | Dispatch is irrevocable; the caller must complete, and requeue is a new enqueue |
@@ -957,6 +982,15 @@ budget is genuinely violated after the one permitted transactional rebase; the
 production state remains unchanged, and oracle state is rolled back so the
 shared trace can continue.
 
+The production regression suite MUST include a public-operation-only trace in
+the unrestricted domain with `D=2`, `maxFlows=70`, `maxLiveJobs=3`, one running
+unit-weight anchor, and 69 consecutive prime weights above `2^60`. Two visits
+of two unit-cost jobs per fractional flow MUST reproduce the bounded numeric
+failure after 273 total successful admissions and MUST verify that the rejected
+enqueue is an atomic no-op. The same prime weights MUST be rejected at
+registration by a divisor-constrained domain whose common scale they do not
+divide.
+
 When no such expected rejection occurs, the reference model and production
 implementation MUST agree on:
 

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

@@ -28,6 +28,8 @@ enum Rejected implements RegisterFlowResult {
         DUPLICATE_REGISTERED_ID,
         /** The configured registration limit is full. */
         FLOW_LIMIT,
+        /** The weight is outside the configured weight domain. */
+        WEIGHT_OUTSIDE_DOMAIN,
         /** The lifetime flow sequence has been exhausted. */
         FLOW_SEQUENCE_EXHAUSTED
     }

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

@@ -20,12 +20,14 @@
  * @param maxLiveJobs maximum queued plus running jobs, in
  *        {@code [depth, Integer.MAX_VALUE]}
  * @param cancellationAccounting virtual fairness accounting policy for cancelled queued jobs
+ * @param weightDomain registration policy for fixed flow weights
  */
 public record SchedulerConfig(
         int depth,
         int maxFlows,
         int maxLiveJobs,
-        CancellationAccounting cancellationAccounting) {
+        CancellationAccounting cancellationAccounting,
+        WeightDomain weightDomain) {
     /**
      * Maximum representable and validated issue depth.
      *
@@ -42,7 +44,24 @@ public record SchedulerConfig(
      * @param maxLiveJobs maximum queued plus running jobs, in {@code [depth, Integer.MAX_VALUE]}
      */
     public SchedulerConfig(int depth, int maxFlows, int maxLiveJobs) {
-        this(depth, maxFlows, maxLiveJobs, CancellationAccounting.CHARGE_RESERVED_COST);
+        this(depth, maxFlows, maxLiveJobs,
+                CancellationAccounting.CHARGE_RESERVED_COST, WeightDomain.unrestricted());
+    }
+
+    /**
+     * Creates a configuration with an unrestricted weight domain.
+     *
+     * @param depth maximum outstanding issue depth, in {@code [1, 1_000_000]}
+     * @param maxFlows maximum simultaneously registered flows, in {@code [1, Integer.MAX_VALUE]}
+     * @param maxLiveJobs maximum queued plus running jobs, in {@code [depth, Integer.MAX_VALUE]}
+     * @param cancellationAccounting virtual fairness accounting policy for cancelled queued jobs
+     */
+    public SchedulerConfig(
+            int depth,
+            int maxFlows,
+            int maxLiveJobs,
+            CancellationAccounting cancellationAccounting) {
+        this(depth, maxFlows, maxLiveJobs, cancellationAccounting, WeightDomain.unrestricted());
     }
 
     /**
@@ -55,13 +74,15 @@ public SchedulerConfig(int depth, int maxFlows, int maxLiveJobs) {
      * @param maxLiveJobs maximum queued plus running jobs, in
      *        {@code [depth, Integer.MAX_VALUE]}
      * @param cancellationAccounting virtual fairness accounting policy for cancelled queued jobs
+     * @param weightDomain registration policy for fixed flow weights
      * @throws IllegalArgumentException if depth is outside {@code [1, 1_000_000]},
      *         maxFlows is outside {@code [1, Integer.MAX_VALUE]}, or
      *         maxLiveJobs is outside {@code [depth, Integer.MAX_VALUE]}
-     * @throws NullPointerException if cancellationAccounting is null
+     * @throws NullPointerException if cancellationAccounting or weightDomain is null
      */
     public SchedulerConfig {
         Objects.requireNonNull(cancellationAccounting, "cancellationAccounting");
+        Objects.requireNonNull(weightDomain, "weightDomain");
         if (depth < 1 || depth > MAX_DEPTH) {
             throw new IllegalArgumentException("depth must be in [1, 1_000_000]");
         }

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

@@ -36,7 +36,9 @@
  * {@code O(queuedJobs + registeredFlows)} and is computed transactionally before it becomes observable. Internal
  * records are bounded by configured live-job and registration limits; terminal tombstones are not retained.
  *
- * <p>Weights and costs are positive {@code long} values in {@code [1, Long.MAX_VALUE]}. Tags are exact reduced
+ * <p>Weights and costs are positive {@code long} values in {@code [1, Long.MAX_VALUE]}. A configured
+ * {@link WeightDomain} may further restrict registered weights to divisors of one common scale, preventing new
+ * denominator factors from accumulating across flows. Tags are exact reduced
  * non-negative rationals: each numerator and denominator retained in scheduler state has bit length at most 4096,
  * and each canonical raw or reduced component of an exact primitive has bit length at most 8193. No rounding or
  * floating-point fallback is permitted. If a newly computed start or finish tag first exceeds the persistent budget,
@@ -86,6 +88,9 @@ public SfqdScheduler(SchedulerConfig config) {
      *
      * <p>A success linearizes at insertion into both registration indexes and sequence advancement. A rejection
      * linearizes at the first applicable check and does not mutate state or consume a sequence.
+     * A divisor-constrained {@link WeightDomain} returns
+     * {@link RegisterFlowResult.Rejected#WEIGHT_OUTSIDE_DOMAIN} before identity, capacity, and sequence checks when
+     * the positive weight does not divide the configured common scale.
      *
      * @param flowId stable non-null flow identifier
      * @param weight fixed registration weight in {@code [1, Long.MAX_VALUE]}
@@ -98,6 +103,9 @@ public RegisterFlowResult registerFlow(F flowId, long weight) {
         requirePositive(weight, "weight");
         lock.lock();
         try {
+            if (!config.weightDomain().permits(weight)) {
+                return RegisterFlowResult.Rejected.WEIGHT_OUTSIDE_DOMAIN;
+            }
             if (registeredById.containsKey(flowId)) {
                 return RegisterFlowResult.Rejected.DUPLICATE_REGISTERED_ID;
             }

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

@@ -0,0 +1,86 @@
+package io.github.pzhin.sfqd;
+
+import java.util.Objects;
+import java.util.OptionalLong;
+
+/**
+ * Immutable registration policy for flow weights.
+ *
+ * <p>{@link #unrestricted()} preserves the full syntactic weight range in
+ * {@code [1, Long.MAX_VALUE]}. It does not constrain the combined arithmetic
+ * structure of registered weights, so an otherwise valid workload can reach
+ * the scheduler's exact-tag representation budget after a bounded number of
+ * admissions.
+ *
+ * <p>{@link #divisorsOf(long)} admits only weights that divide one fixed common
+ * scale. Every reduced {@code cost / weight} denominator then divides that
+ * scale, and addition, maximum, and rebase subtraction cannot introduce a new
+ * denominator factor. This prevents pairwise-coprime weights from causing
+ * denominator growth. It is not a promise that {@code NUMERIC_LIMIT} is
+ * impossible: numerators and accumulated cancellation debt remain bounded by
+ * the scheduler's documented representation budget.
+ */
+public final class WeightDomain {
+    private static final WeightDomain UNRESTRICTED = new WeightDomain(OptionalLong.empty());
+
+    private final OptionalLong commonScale;
+
+    private WeightDomain(OptionalLong commonScale) {
+        this.commonScale = commonScale;
+    }
+
+    /**
+     * Returns the compatibility policy accepting every positive {@code long} weight.
+     *
+     * @return unrestricted weight domain
+     */
+    public static WeightDomain unrestricted() {
+        return UNRESTRICTED;
+    }
+
+    /**
+     * Returns a domain accepting exactly the positive divisors of {@code commonScale}.
+     *
+     * @param commonScale fixed positive common weight scale
+     * @return divisor-constrained weight domain
+     * @throws IllegalArgumentException if commonScale is not positive
+     */
+    public static WeightDomain divisorsOf(long commonScale) {
+        if (commonScale <= 0L) {
+            throw new IllegalArgumentException("commonScale must be positive");
+        }
+        return new WeightDomain(OptionalLong.of(commonScale));
+    }
+
+    /**
+     * Returns the common scale when weights are divisor-constrained.
+     *
+     * @return common scale, or empty for the unrestricted domain
+     */
+    public OptionalLong commonScale() {
+        return commonScale;
+    }
+
+    boolean permits(long weight) {
+        return commonScale.isEmpty() || commonScale.getAsLong() % weight == 0L;
+    }
+
+    @Override
+    public boolean equals(Object other) {
+        return this == other
+                || other instanceof WeightDomain domain
+                && commonScale.equals(domain.commonScale);
+    }
+
+    @Override
+    public int hashCode() {
+        return Objects.hash(commonScale);
+    }
+
+    @Override
+    public String toString() {
+        return commonScale.isEmpty()
+                ? "WeightDomain[unrestricted]"
+                : "WeightDomain[divisorsOf=" + commonScale.getAsLong() + "]";
+    }
+}