From 29e30183a211025c19e4439f8a1e042eb3efb615 Mon Sep 17 00:00:00 2001 From: Andrea Vibelli Date: Sun, 11 Oct 2026 00:38:52 +0200 Subject: [PATCH 1/4] Introduce RunningTracker SPI for the queue running counter (capacity PoC, PR1) Extract the per-queue running-counter logic behind a RunningTracker SPI and refactor QueueManagerImpl to use it. Behaviour-neutral: CasRunningTracker keeps the existing optimistic compare-and-swap semantics (getMetadataValue / replaceValue, throwing ConcurrentModificationException on version conflict). This is the first step towards replacing the single-key counter contention with an Infinispan clustered StrongCounter, to scale concurrency without adding a new system. No runtime behaviour change yet. Adds CasRunningTrackerTest (pure JUnit + Mockito, no Infinispan/containers). Co-Authored-By: Claude Opus 4.8 --- .../jboss/pnc/rex/core/QueueManagerImpl.java | 71 ++----- .../rex/core/counter/CasRunningTracker.java | 104 ++++++++++ .../pnc/rex/core/counter/RunningTracker.java | 71 +++++++ .../test/counter/CasRunningTrackerTest.java | 190 ++++++++++++++++++ 4 files changed, 381 insertions(+), 55 deletions(-) create mode 100644 core/src/main/java/org/jboss/pnc/rex/core/counter/CasRunningTracker.java create mode 100644 core/src/main/java/org/jboss/pnc/rex/core/counter/RunningTracker.java create mode 100644 core/src/test/java/org/jboss/pnc/rex/test/counter/CasRunningTrackerTest.java diff --git a/core/src/main/java/org/jboss/pnc/rex/core/QueueManagerImpl.java b/core/src/main/java/org/jboss/pnc/rex/core/QueueManagerImpl.java index 81beec2..762c57c 100644 --- a/core/src/main/java/org/jboss/pnc/rex/core/QueueManagerImpl.java +++ b/core/src/main/java/org/jboss/pnc/rex/core/QueueManagerImpl.java @@ -13,14 +13,15 @@ import org.jboss.pnc.rex.core.api.TaskRegistry; import org.jboss.pnc.rex.core.counter.Counter; import org.jboss.pnc.rex.core.counter.MaxConcurrent; -import org.jboss.pnc.rex.core.counter.Running; +import org.jboss.pnc.rex.core.counter.RunningTracker; import org.jboss.pnc.rex.core.delegates.FaultToleranceDecorator; import org.jboss.pnc.rex.model.Task; import jakarta.enterprise.context.ApplicationScoped; import jakarta.transaction.Transactional; -import java.util.*; +import java.util.List; +import java.util.Map; import java.util.stream.Collectors; import static jakarta.transaction.Transactional.TxType.MANDATORY; @@ -32,13 +33,13 @@ public class QueueManagerImpl implements QueueManager { private static final String DEFAULT_QUEUE_NAMING = "DEFAULT"; private final Counter max; - private final Counter running; + private final RunningTracker running; private final TaskRegistry container; private final TaskController controller; private final FaultToleranceDecorator ft; public QueueManagerImpl(@MaxConcurrent Counter max, - @Running Counter running, + RunningTracker running, TaskRegistry container, TaskController controller, FaultToleranceDecorator ft) { @@ -54,27 +55,18 @@ public QueueManagerImpl(@MaxConcurrent Counter max, public void poke() { log.info("QUEUE: Poking Task queues"); Map maxEntries = max.entries(); - Map runningEntries = running.entries(); - Set consideredQueues = new HashSet<>(); for (var queue : maxEntries.keySet()) { Long maxValue = maxEntries.get(queue); - Long runningValue = runningEntries.get(queue); + long runningValue = running.get(queue); if (runningValue >= maxValue) { log.debug("QUEUE '{}': Maximum number of parallel tasks reached.({} out of {})", queue == null ? DEFAULT_QUEUE_NAMING : queue, runningValue, maxValue); - } else { - consideredQueues.add(queue); + continue; } - } - - for (var queue : consideredQueues) { - Long maxValue = maxEntries.get(queue); - VersionedValue runningMetadata = running.getMetadataValue(queue); - Long runningValue = runningMetadata.getValue(); long freeSpace = maxValue - runningValue; List randomEnqueuedTasks = container.getEnqueuedTasksByQueueName(queue, freeSpace); @@ -91,34 +83,18 @@ public void poke() { randomEnqueuedTasks.forEach(task -> controller.dequeue(task.getName())); - log.info("QUEUE '{}': Increasing running counter. ({} to {}) [ISPN-VERSION:{}]", + long newRunning = running.reserve(queue, randomEnqueuedTasks.size()); + log.info("QUEUE '{}': Increased running counter to {}.", queue == null ? DEFAULT_QUEUE_NAMING : queue, - runningValue, - (runningValue + randomEnqueuedTasks.size()), - runningMetadata.getVersion()); - if (!running.replaceValue(queue, runningMetadata, runningValue + randomEnqueuedTasks.size())) { - RuntimeException e = new ConcurrentModificationException("Running counter was modified concurrently."); - log.error("QUEUE '{}': Concurrent modification detected.", queue == null ? DEFAULT_QUEUE_NAMING : queue, e); - throw e; - } + newRunning); } } @Override @Transactional(MANDATORY) public void decreaseRunningCounter(@Nullable String name) { - VersionedValue runningMetadata = running.getMetadataValue(name); - long runningValue = runningMetadata.getValue() - 1; - log.info("QUEUE '{}': Decreasing running counter by one. ({} to {}) [ISPN-VERSION:{}]", - name == null ? DEFAULT_QUEUE_NAMING : name, - runningMetadata.getValue(), - runningValue, - runningMetadata.getVersion()); - if (!running.replaceValue(name, runningMetadata, runningValue)) { - RuntimeException e = new ConcurrentModificationException("Running counter was modified concurrently."); - log.error("QUEUE: Concurrent modification detected.", e); - throw e; - } + running.decrement(name); + log.info("QUEUE '{}': Decreased running counter by one.", name == null ? DEFAULT_QUEUE_NAMING : name); } @Override @@ -146,8 +122,7 @@ public Long getMaximumConcurrency(@Nullable String name) { @Override public Long getRunningCounter(@Nullable String name) { - VersionedValue meta = running.getMetadataValue(name); - return meta == null ? null : meta.getValue(); + return running.peek(name); } @Override @@ -157,24 +132,10 @@ public void synchronizeRunningCounter() { .stream() .collect(groupingBy(Task::getQueue)); - - Set existingQueues = running.entries().keySet(); - for (String queue : existingQueues) { - var runningValue = running.getMetadataValue(queue); + for (String queue : running.queues()) { var tasksInQueue = tasksByQueue.get(queue); - - long actualValue; - if (tasksInQueue == null) { - actualValue = 0L; - } else { - actualValue = tasksInQueue.size(); - } - - if (!runningValue.getValue().equals(actualValue)) { - log.info("Synchronizing running counter. Mismatch between active tasks and counter found. Previous value '{}' -> new value '{}'", runningValue.getValue(), actualValue); - running.replaceValue(queue, runningValue, actualValue); - } - + long actualValue = tasksInQueue == null ? 0L : tasksInQueue.size(); + running.reconcile(queue, actualValue); } } diff --git a/core/src/main/java/org/jboss/pnc/rex/core/counter/CasRunningTracker.java b/core/src/main/java/org/jboss/pnc/rex/core/counter/CasRunningTracker.java new file mode 100644 index 0000000..491be42 --- /dev/null +++ b/core/src/main/java/org/jboss/pnc/rex/core/counter/CasRunningTracker.java @@ -0,0 +1,104 @@ +/* + * SPDX-FileCopyrightText: Copyright © 2021 Red Hat, Inc., and individual contributors as indicated by the @author tags. + * SPDX-License-Identifier: Apache-2.0 + */ +package org.jboss.pnc.rex.core.counter; + +import jakarta.annotation.Nullable; +import jakarta.enterprise.context.ApplicationScoped; +import lombok.extern.slf4j.Slf4j; +import org.infinispan.client.hotrod.VersionedValue; + +import java.util.ConcurrentModificationException; +import java.util.Set; + +/** + * Optimistic compare-and-swap implementation of {@link RunningTracker}, backed by the existing + * {@code rex-counter} cache through the {@link Running} {@link Counter}. + *

+ * This preserves the historical behavior: every mutation reads the versioned value and writes it + * back with {@code replaceWithVersion}; on a version conflict it throws + * {@link ConcurrentModificationException} so the surrounding {@code internal-retry} guarded + * transaction is replayed. + */ +@Slf4j +@ApplicationScoped +public class CasRunningTracker implements RunningTracker { + + private static final String DEFAULT_QUEUE_NAMING = "DEFAULT"; + + private final Counter counter; + + public CasRunningTracker(@Running Counter counter) { + this.counter = counter; + } + + @Override + public long get(@Nullable String queue) { + VersionedValue meta = counter.getMetadataValue(queue); + return meta == null ? 0L : meta.getValue(); + } + + @Override + @Nullable + public Long peek(@Nullable String queue) { + VersionedValue meta = counter.getMetadataValue(queue); + return meta == null ? null : meta.getValue(); + } + + @Override + public long reserve(@Nullable String queue, long n) { + VersionedValue meta = counter.getMetadataValue(queue); + long newValue = meta.getValue() + n; + if (!counter.replaceValue(queue, meta, newValue)) { + throw concurrentModification(queue); + } + return newValue; + } + + @Override + public void release(@Nullable String queue, long n) { + VersionedValue meta = counter.getMetadataValue(queue); + long newValue = meta.getValue() - n; + if (!counter.replaceValue(queue, meta, newValue)) { + throw concurrentModification(queue); + } + } + + @Override + public void decrement(@Nullable String queue) { + release(queue, 1); + } + + @Override + public void reconcile(@Nullable String queue, long actual) { + VersionedValue meta = counter.getMetadataValue(queue); + if (meta == null) { + counter.initialize(queue, actual); + return; + } + if (meta.getValue() != actual) { + log.info("QUEUE '{}': Synchronizing running counter. Mismatch between active tasks and counter found. " + + "Previous value '{}' -> new value '{}'", + queue == null ? DEFAULT_QUEUE_NAMING : queue, meta.getValue(), actual); + counter.replaceValue(queue, meta, actual); + } + } + + @Override + public void initialize(@Nullable String queue, long value) { + counter.initialize(queue, value); + } + + @Override + public Set queues() { + return counter.entries().keySet(); + } + + private ConcurrentModificationException concurrentModification(String queue) { + ConcurrentModificationException e = + new ConcurrentModificationException("Running counter was modified concurrently."); + log.error("QUEUE '{}': Concurrent modification detected.", queue == null ? DEFAULT_QUEUE_NAMING : queue, e); + return e; + } +} diff --git a/core/src/main/java/org/jboss/pnc/rex/core/counter/RunningTracker.java b/core/src/main/java/org/jboss/pnc/rex/core/counter/RunningTracker.java new file mode 100644 index 0000000..8a830f1 --- /dev/null +++ b/core/src/main/java/org/jboss/pnc/rex/core/counter/RunningTracker.java @@ -0,0 +1,71 @@ +/* + * SPDX-FileCopyrightText: Copyright © 2021 Red Hat, Inc., and individual contributors as indicated by the @author tags. + * SPDX-License-Identifier: Apache-2.0 + */ +package org.jboss.pnc.rex.core.counter; + +import jakarta.annotation.Nullable; + +import java.util.Set; + +/** + * Tracks the number of currently running Tasks per queue (the "running counter"). + *

+ * This is an abstraction over the storage/concurrency mechanism of the running counter so the + * implementation can be swapped (e.g. optimistic compare-and-swap on a cache entry vs. an + * Infinispan clustered {@code StrongCounter}) without touching the {@code QueueManager} logic. + *

+ * A {@code null} queue name refers to the DEFAULT queue. + */ +public interface RunningTracker { + + /** + * Current number of running Tasks in the queue. For the DEFAULT queue the counter is lazily + * initialized to {@code 0} if it does not exist yet. + */ + long get(@Nullable String queue); + + /** + * Current value without lazily creating it. Returns {@code null} if the queue's counter does + * not exist (used by the admin/REST getter to distinguish a missing named queue). + */ + @Nullable + Long peek(@Nullable String queue); + + /** + * Atomically add {@code n} to the running counter (reserve capacity for {@code n} starting + * Tasks) and return the new value. + * + * @throws java.util.ConcurrentModificationException if the counter was modified concurrently + * (CAS-based implementations use this to trigger the outer transaction retry) + */ + long reserve(@Nullable String queue, long n); + + /** + * Atomically subtract {@code n} from the running counter (compensation for capacity that was + * reserved but not consumed). + */ + void release(@Nullable String queue, long n); + + /** + * Decrease the running counter by one. Invoked when a Task leaves the RUNNING state group. + */ + void decrement(@Nullable String queue); + + /** + * Set the running counter of the queue to {@code actual} if it currently differs (drift + * reconciliation against the real number of running Tasks). + */ + void reconcile(@Nullable String queue, long actual); + + /** + * Initialize the running counter of a (new) queue to {@code value}. + */ + void initialize(@Nullable String queue, long value); + + /** + * Names of all known queues that have a running counter. The DEFAULT queue is represented by + * {@code null}. + */ + Set queues(); +} diff --git a/core/src/test/java/org/jboss/pnc/rex/test/counter/CasRunningTrackerTest.java b/core/src/test/java/org/jboss/pnc/rex/test/counter/CasRunningTrackerTest.java new file mode 100644 index 0000000..2fa39f8 --- /dev/null +++ b/core/src/test/java/org/jboss/pnc/rex/test/counter/CasRunningTrackerTest.java @@ -0,0 +1,190 @@ +/* + * SPDX-FileCopyrightText: Copyright © 2021 Red Hat, Inc., and individual contributors as indicated by the @author tags. + * SPDX-License-Identifier: Apache-2.0 + */ +package org.jboss.pnc.rex.test.counter; + +import org.infinispan.client.hotrod.VersionedValue; +import org.jboss.pnc.rex.core.counter.CasRunningTracker; +import org.jboss.pnc.rex.core.counter.Counter; +import org.junit.jupiter.api.Test; + +import java.util.ConcurrentModificationException; +import java.util.LinkedHashMap; +import java.util.Map; + +import static org.assertj.core.api.Assertions.assertThat; +import static org.assertj.core.api.Assertions.assertThatThrownBy; +import static org.mockito.ArgumentMatchers.any; +import static org.mockito.ArgumentMatchers.anyString; +import static org.mockito.ArgumentMatchers.eq; +import static org.mockito.Mockito.mock; +import static org.mockito.Mockito.never; +import static org.mockito.Mockito.verify; +import static org.mockito.Mockito.when; + +/** + * Pure unit test (no Quarkus/Infinispan) for {@link CasRunningTracker}: verifies it maps the + * high-level {@link org.jboss.pnc.rex.core.counter.RunningTracker} operations onto the underlying + * optimistic compare-and-swap {@link Counter} primitives, including conflict handling. + */ +class CasRunningTrackerTest { + + private static VersionedValue vv(long value, long version) { + return new VersionedValue<>() { + @Override + public long getVersion() { + return version; + } + + @Override + public Long getValue() { + return value; + } + }; + } + + @Test + void getReturnsCurrentValue() { + Counter counter = mock(Counter.class); + when(counter.getMetadataValue("q")).thenReturn(vv(3L, 1L)); + + assertThat(new CasRunningTracker(counter).get("q")).isEqualTo(3L); + } + + @Test + void getReturnsZeroWhenCounterMissing() { + Counter counter = mock(Counter.class); + when(counter.getMetadataValue("q")).thenReturn(null); + + assertThat(new CasRunningTracker(counter).get("q")).isZero(); + } + + @Test + void peekReturnsNullWhenCounterMissing() { + Counter counter = mock(Counter.class); + when(counter.getMetadataValue(null)).thenReturn(null); + + assertThat(new CasRunningTracker(counter).peek(null)).isNull(); + } + + @Test + void peekReturnsValueWhenPresent() { + Counter counter = mock(Counter.class); + when(counter.getMetadataValue("q")).thenReturn(vv(7L, 2L)); + + assertThat(new CasRunningTracker(counter).peek("q")).isEqualTo(7L); + } + + @Test + void reserveAddsAmountAndReturnsNewValue() { + Counter counter = mock(Counter.class); + VersionedValue meta = vv(2L, 5L); + when(counter.getMetadataValue("q")).thenReturn(meta); + when(counter.replaceValue("q", meta, 5L)).thenReturn(true); // 2 + 3 + + long result = new CasRunningTracker(counter).reserve("q", 3); + + assertThat(result).isEqualTo(5L); + verify(counter).replaceValue("q", meta, 5L); + } + + @Test + void reserveThrowsConcurrentModificationOnVersionConflict() { + Counter counter = mock(Counter.class); + VersionedValue meta = vv(2L, 5L); + when(counter.getMetadataValue("q")).thenReturn(meta); + when(counter.replaceValue(eq("q"), eq(meta), any())).thenReturn(false); + + assertThatThrownBy(() -> new CasRunningTracker(counter).reserve("q", 1)) + .isInstanceOf(ConcurrentModificationException.class); + } + + @Test + void releaseSubtractsAmount() { + Counter counter = mock(Counter.class); + VersionedValue meta = vv(5L, 9L); + when(counter.getMetadataValue("q")).thenReturn(meta); + when(counter.replaceValue("q", meta, 2L)).thenReturn(true); // 5 - 3 + + new CasRunningTracker(counter).release("q", 3); + + verify(counter).replaceValue("q", meta, 2L); + } + + @Test + void decrementSubtractsOne() { + Counter counter = mock(Counter.class); + VersionedValue meta = vv(5L, 9L); + when(counter.getMetadataValue("q")).thenReturn(meta); + when(counter.replaceValue("q", meta, 4L)).thenReturn(true); + + new CasRunningTracker(counter).decrement("q"); + + verify(counter).replaceValue("q", meta, 4L); + } + + @Test + void decrementThrowsConcurrentModificationOnVersionConflict() { + Counter counter = mock(Counter.class); + VersionedValue meta = vv(5L, 9L); + when(counter.getMetadataValue("q")).thenReturn(meta); + when(counter.replaceValue(eq("q"), eq(meta), any())).thenReturn(false); + + assertThatThrownBy(() -> new CasRunningTracker(counter).decrement("q")) + .isInstanceOf(ConcurrentModificationException.class); + } + + @Test + void reconcileWritesActualWhenDifferent() { + Counter counter = mock(Counter.class); + VersionedValue meta = vv(5L, 9L); + when(counter.getMetadataValue("q")).thenReturn(meta); + when(counter.replaceValue("q", meta, 2L)).thenReturn(true); + + new CasRunningTracker(counter).reconcile("q", 2); + + verify(counter).replaceValue("q", meta, 2L); + } + + @Test + void reconcileIsNoopWhenAlreadyEqual() { + Counter counter = mock(Counter.class); + VersionedValue meta = vv(4L, 9L); + when(counter.getMetadataValue("q")).thenReturn(meta); + + new CasRunningTracker(counter).reconcile("q", 4); + + verify(counter, never()).replaceValue(anyString(), any(VersionedValue.class), any()); + } + + @Test + void reconcileInitializesWhenCounterMissing() { + Counter counter = mock(Counter.class); + when(counter.getMetadataValue("q")).thenReturn(null); + + new CasRunningTracker(counter).reconcile("q", 6); + + verify(counter).initialize("q", 6L); + } + + @Test + void initializeDelegatesToCounter() { + Counter counter = mock(Counter.class); + + new CasRunningTracker(counter).initialize("q", 0L); + + verify(counter).initialize("q", 0L); + } + + @Test + void queuesReturnsCounterKeysIncludingDefaultNull() { + Counter counter = mock(Counter.class); + Map entries = new LinkedHashMap<>(); + entries.put(null, 1L); // DEFAULT queue + entries.put("vm", 2L); + when(counter.entries()).thenReturn(entries); + + assertThat(new CasRunningTracker(counter).queues()).containsExactlyInAnyOrder(null, "vm"); + } +} From 80909014547fb46afc961bb81698031cf8157471 Mon Sep 17 00:00:00 2001 From: Andrea Vibelli Date: Sun, 11 Oct 2026 01:14:27 +0200 Subject: [PATCH 2/4] Add StrongCounter running-counter tracker behind a flag (capacity PoC, PR2) Add an Infinispan clustered StrongCounter implementation of RunningTracker to remove the single-key compare-and-swap contention / pessimistic-lock timeouts of the cache-entry counter, which is the main scheduler bottleneck at high concurrency (and already a cost on the current 2-replica setup). - StrongCounterRunningTracker: one unbounded StrongCounter per queue (RUNNING / RUNNING-), atomic increments/decrements, non-negative floor, queue discovery via CounterManager#getCounterNames. - CounterManagerProducer: exposes the CounterManager from the Hot Rod RemoteCacheManager (same cluster; no new system). - RunningTrackerProducer: selects cas (default) or strong via rex.queue.running-tracker, so this ships dark and is flipped per environment. - QueueManagerImpl.poke(): atomic counters use reserve-then-compensate before dequeue (prevents cross-node over-admission); the cas path is unchanged. Decrement stays in-transaction and drift is healed by reconciliation. - synchronizeRunningCounter(): reconcile the union of max-counter queues and running-counter queues (restart-safe for volatile StrongCounters), null-safe grouping for the DEFAULT queue. - RunningCounterStartupReconciler: on startup, rebuild volatile running counters from live task state when strong is active. Tests: StrongCounterRunningTrackerTest (unit, mocked), and StrongCounterQueueTest reruns the full QueueTest suite under rex.queue.running-tracker=strong. Co-Authored-By: Claude Opus 4.8 --- .../jboss/pnc/rex/core/QueueManagerImpl.java | 131 +++++++++++--- .../rex/core/counter/CasRunningTracker.java | 12 +- .../core/counter/CounterManagerProducer.java | 26 +++ .../RunningCounterStartupReconciler.java | 49 +++++ .../pnc/rex/core/counter/RunningTracker.java | 12 ++ .../core/counter/RunningTrackerProducer.java | 45 +++++ .../counter/StrongCounterRunningTracker.java | 155 ++++++++++++++++ core/src/main/resources/application.yaml | 7 + .../pnc/rex/test/StrongCounterQueueTest.java | 32 ++++ .../StrongCounterRunningTrackerTest.java | 170 ++++++++++++++++++ 10 files changed, 608 insertions(+), 31 deletions(-) create mode 100644 core/src/main/java/org/jboss/pnc/rex/core/counter/CounterManagerProducer.java create mode 100644 core/src/main/java/org/jboss/pnc/rex/core/counter/RunningCounterStartupReconciler.java create mode 100644 core/src/main/java/org/jboss/pnc/rex/core/counter/RunningTrackerProducer.java create mode 100644 core/src/main/java/org/jboss/pnc/rex/core/counter/StrongCounterRunningTracker.java create mode 100644 core/src/test/java/org/jboss/pnc/rex/test/StrongCounterQueueTest.java create mode 100644 core/src/test/java/org/jboss/pnc/rex/test/counter/StrongCounterRunningTrackerTest.java diff --git a/core/src/main/java/org/jboss/pnc/rex/core/QueueManagerImpl.java b/core/src/main/java/org/jboss/pnc/rex/core/QueueManagerImpl.java index 762c57c..72683b5 100644 --- a/core/src/main/java/org/jboss/pnc/rex/core/QueueManagerImpl.java +++ b/core/src/main/java/org/jboss/pnc/rex/core/QueueManagerImpl.java @@ -20,12 +20,15 @@ import jakarta.enterprise.context.ApplicationScoped; import jakarta.transaction.Transactional; +import java.util.ArrayList; +import java.util.HashMap; +import java.util.HashSet; import java.util.List; import java.util.Map; +import java.util.Set; import java.util.stream.Collectors; import static jakarta.transaction.Transactional.TxType.MANDATORY; -import static java.util.stream.Collectors.groupingBy; @Slf4j @ApplicationScoped @@ -58,35 +61,98 @@ public void poke() { for (var queue : maxEntries.keySet()) { Long maxValue = maxEntries.get(queue); - long runningValue = running.get(queue); - - if (runningValue >= maxValue) { - log.debug("QUEUE '{}': Maximum number of parallel tasks reached.({} out of {})", - queue == null ? DEFAULT_QUEUE_NAMING : queue, - runningValue, - maxValue); - continue; - } - - long freeSpace = maxValue - runningValue; - List randomEnqueuedTasks = container.getEnqueuedTasksByQueueName(queue, freeSpace); - if (randomEnqueuedTasks.isEmpty()) { - continue; + if (running.requiresReservation()) { + pokeWithReservation(queue, maxValue); + } else { + pokeLegacy(queue, maxValue); } + } + } - log.info("QUEUE '{}': Free space of {} found. Scheduling {} task(s) of {}", + /** + * Historical order for optimistic compare-and-swap counters: dequeue first, then increment. + * The shared-key version check serializes admission across nodes (a stale increment fails and + * the whole transaction is retried), so no over-admission can happen. + */ + private void pokeLegacy(String queue, Long maxValue) { + long runningValue = running.get(queue); + if (runningValue >= maxValue) { + log.debug("QUEUE '{}': Maximum number of parallel tasks reached.({} out of {})", queue == null ? DEFAULT_QUEUE_NAMING : queue, - freeSpace, - randomEnqueuedTasks.size(), - randomEnqueuedTasks.stream().map(Task::getName).collect(Collectors.toList()) - ); + runningValue, + maxValue); + return; + } + + long freeSpace = maxValue - runningValue; + List randomEnqueuedTasks = container.getEnqueuedTasksByQueueName(queue, freeSpace); + if (randomEnqueuedTasks.isEmpty()) { + return; + } + + log.info("QUEUE '{}': Free space of {} found. Scheduling {} task(s) of {}", + queue == null ? DEFAULT_QUEUE_NAMING : queue, + freeSpace, + randomEnqueuedTasks.size(), + randomEnqueuedTasks.stream().map(Task::getName).collect(Collectors.toList()) + ); + + randomEnqueuedTasks.forEach(task -> controller.dequeue(task.getName())); - randomEnqueuedTasks.forEach(task -> controller.dequeue(task.getName())); + long newRunning = running.reserve(queue, randomEnqueuedTasks.size()); + log.info("QUEUE '{}': Increased running counter to {}.", + queue == null ? DEFAULT_QUEUE_NAMING : queue, + newRunning); + } - long newRunning = running.reserve(queue, randomEnqueuedTasks.size()); - log.info("QUEUE '{}': Increased running counter to {}.", + /** + * Order for atomic, non-transactional counters (e.g. StrongCounter): reserve capacity first, + * release any overshoot, then dequeue only the amount we are allowed to start. This prevents + * two nodes from both observing the same free space and over-admitting. + */ + private void pokeWithReservation(String queue, Long maxValue) { + long runningValue = running.get(queue); + if (runningValue >= maxValue) { + log.debug("QUEUE '{}': Maximum number of parallel tasks reached.({} out of {})", queue == null ? DEFAULT_QUEUE_NAMING : queue, - newRunning); + runningValue, + maxValue); + return; + } + + long freeSpace = maxValue - runningValue; + List candidates = container.getEnqueuedTasksByQueueName(queue, freeSpace); + if (candidates.isEmpty()) { + return; + } + + int want = candidates.size(); + long afterReserve = running.reserve(queue, want); + int allowed = want; + if (afterReserve > maxValue) { + long giveBack = Math.min(afterReserve - maxValue, want); + running.release(queue, giveBack); + allowed = (int) (want - giveBack); + } + if (allowed <= 0) { + return; + } + + List toSchedule = candidates.subList(0, allowed); + log.info("QUEUE '{}': Reserved {} slot(s). Scheduling {} task(s) of {}", + queue == null ? DEFAULT_QUEUE_NAMING : queue, + allowed, + toSchedule.size(), + toSchedule.stream().map(Task::getName).collect(Collectors.toList()) + ); + + try { + toSchedule.forEach(task -> controller.dequeue(task.getName())); + } catch (RuntimeException e) { + // The transaction rolls back every dequeue done in this invocation; release the whole + // reservation so the (non-transactional) counter does not drift. + running.release(queue, allowed); + throw e; } } @@ -128,12 +194,19 @@ public Long getRunningCounter(@Nullable String name) { @Override @Transactional(MANDATORY) public void synchronizeRunningCounter() { - Map> tasksByQueue = container.getTasks(false, false, true, false, false, null) - .stream() - .collect(groupingBy(Task::getQueue)); + // null-safe grouping: DEFAULT queue tasks have a null queue, which Collectors.groupingBy rejects + Map> tasksByQueue = new HashMap<>(); + for (Task task : container.getTasks(false, false, true, false, false, null)) { + tasksByQueue.computeIfAbsent(task.getQueue(), q -> new ArrayList<>()).add(task); + } + + // reconcile every known queue. Max counters survive a restart (persistent cache), while the + // running counters may not (e.g. volatile StrongCounters), so union both sources. + Set queues = new HashSet<>(max.entries().keySet()); + queues.addAll(running.queues()); - for (String queue : running.queues()) { - var tasksInQueue = tasksByQueue.get(queue); + for (String queue : queues) { + List tasksInQueue = tasksByQueue.get(queue); long actualValue = tasksInQueue == null ? 0L : tasksInQueue.size(); running.reconcile(queue, actualValue); } diff --git a/core/src/main/java/org/jboss/pnc/rex/core/counter/CasRunningTracker.java b/core/src/main/java/org/jboss/pnc/rex/core/counter/CasRunningTracker.java index 491be42..d9aa5de 100644 --- a/core/src/main/java/org/jboss/pnc/rex/core/counter/CasRunningTracker.java +++ b/core/src/main/java/org/jboss/pnc/rex/core/counter/CasRunningTracker.java @@ -5,7 +5,6 @@ package org.jboss.pnc.rex.core.counter; import jakarta.annotation.Nullable; -import jakarta.enterprise.context.ApplicationScoped; import lombok.extern.slf4j.Slf4j; import org.infinispan.client.hotrod.VersionedValue; @@ -20,9 +19,11 @@ * back with {@code replaceWithVersion}; on a version conflict it throws * {@link ConcurrentModificationException} so the surrounding {@code internal-retry} guarded * transaction is replayed. + *

+ * Instantiated by {@link RunningTrackerProducer} (not a CDI bean itself) so the active + * implementation can be selected by configuration. */ @Slf4j -@ApplicationScoped public class CasRunningTracker implements RunningTracker { private static final String DEFAULT_QUEUE_NAMING = "DEFAULT"; @@ -95,6 +96,13 @@ public Set queues() { return counter.entries().keySet(); } + @Override + public boolean requiresReservation() { + // Optimistic CAS on the shared counter key already serializes admission across nodes; + // keep the historical dequeue-then-increment order. + return false; + } + private ConcurrentModificationException concurrentModification(String queue) { ConcurrentModificationException e = new ConcurrentModificationException("Running counter was modified concurrently."); diff --git a/core/src/main/java/org/jboss/pnc/rex/core/counter/CounterManagerProducer.java b/core/src/main/java/org/jboss/pnc/rex/core/counter/CounterManagerProducer.java new file mode 100644 index 0000000..46faf08 --- /dev/null +++ b/core/src/main/java/org/jboss/pnc/rex/core/counter/CounterManagerProducer.java @@ -0,0 +1,26 @@ +/* + * SPDX-FileCopyrightText: Copyright © 2021 Red Hat, Inc., and individual contributors as indicated by the @author tags. + * SPDX-License-Identifier: Apache-2.0 + */ +package org.jboss.pnc.rex.core.counter; + +import jakarta.enterprise.context.ApplicationScoped; +import jakarta.enterprise.inject.Produces; +import org.infinispan.client.hotrod.RemoteCacheManager; +import org.infinispan.client.hotrod.RemoteCounterManagerFactory; +import org.infinispan.counter.api.CounterManager; + +/** + * Exposes the Infinispan {@link CounterManager} (clustered counters) from the Hot Rod + * {@link RemoteCacheManager} that Quarkus produces for the default Infinispan client. The counter + * subsystem lives inside the same Infinispan cluster — no additional deployment is required. + */ +@ApplicationScoped +public class CounterManagerProducer { + + @Produces + @ApplicationScoped + public CounterManager counterManager(RemoteCacheManager cacheManager) { + return RemoteCounterManagerFactory.asCounterManager(cacheManager); + } +} diff --git a/core/src/main/java/org/jboss/pnc/rex/core/counter/RunningCounterStartupReconciler.java b/core/src/main/java/org/jboss/pnc/rex/core/counter/RunningCounterStartupReconciler.java new file mode 100644 index 0000000..7eac71c --- /dev/null +++ b/core/src/main/java/org/jboss/pnc/rex/core/counter/RunningCounterStartupReconciler.java @@ -0,0 +1,49 @@ +/* + * SPDX-FileCopyrightText: Copyright © 2021 Red Hat, Inc., and individual contributors as indicated by the @author tags. + * SPDX-License-Identifier: Apache-2.0 + */ +package org.jboss.pnc.rex.core.counter; + +import io.quarkus.narayana.jta.QuarkusTransaction; +import io.quarkus.runtime.Startup; +import jakarta.enterprise.context.ApplicationScoped; +import jakarta.enterprise.inject.spi.ObserverMethod; +import lombok.extern.slf4j.Slf4j; +import org.eclipse.microprofile.config.inject.ConfigProperty; +import org.jboss.pnc.rex.core.api.QueueManager; +import org.jboss.pnc.rex.core.delegates.WithRetries; + +/** + * When the StrongCounter tracker is active, reconcile the (volatile) running counters from live task + * state at startup. This restores correct values after a full-cluster restart that dropped the + * counters while tasks survived in the persistent task cache. No-op for the default CAS tracker, + * whose counters live in the persistent cache and survive restarts on their own. + */ +@Slf4j +@ApplicationScoped +public class RunningCounterStartupReconciler { + + private final String impl; + private final QueueManager queueManager; + + public RunningCounterStartupReconciler( + @ConfigProperty(name = RunningTrackerProducer.CONFIG_KEY, defaultValue = RunningTrackerProducer.CAS) String impl, + @WithRetries QueueManager queueManager) { + this.impl = impl; + this.queueManager = queueManager; + } + + @Startup(ObserverMethod.DEFAULT_PRIORITY + 2) + void reconcileOnStartup() { + if (!RunningTrackerProducer.STRONG.equalsIgnoreCase(impl)) { + return; + } + try { + log.info("QUEUE: Reconciling running counters from task state on startup"); + QuarkusTransaction.requiringNew().run(queueManager::synchronizeRunningCounter); + } catch (RuntimeException e) { + log.warn("QUEUE: Startup running-counter reconciliation failed; it will be corrected by " + + "on-demand/maintenance reconciliation", e); + } + } +} diff --git a/core/src/main/java/org/jboss/pnc/rex/core/counter/RunningTracker.java b/core/src/main/java/org/jboss/pnc/rex/core/counter/RunningTracker.java index 8a830f1..9a4eb39 100644 --- a/core/src/main/java/org/jboss/pnc/rex/core/counter/RunningTracker.java +++ b/core/src/main/java/org/jboss/pnc/rex/core/counter/RunningTracker.java @@ -68,4 +68,16 @@ public interface RunningTracker { * {@code null}. */ Set queues(); + + /** + * Whether capacity must be reserved before dequeuing (reserve-then-compensate), as + * opposed to dequeuing first and then incrementing. + *

+ * Non-transactional atomic implementations (e.g. Infinispan StrongCounter) return {@code true}: + * two nodes can otherwise both see the same free space and over-admit, so capacity is reserved + * atomically first and any overshoot is released. Optimistic compare-and-swap implementations + * return {@code false}: the shared-key version check already serializes admission across nodes, + * so the historical dequeue-then-increment order is kept. + */ + boolean requiresReservation(); } diff --git a/core/src/main/java/org/jboss/pnc/rex/core/counter/RunningTrackerProducer.java b/core/src/main/java/org/jboss/pnc/rex/core/counter/RunningTrackerProducer.java new file mode 100644 index 0000000..d3e9925 --- /dev/null +++ b/core/src/main/java/org/jboss/pnc/rex/core/counter/RunningTrackerProducer.java @@ -0,0 +1,45 @@ +/* + * SPDX-FileCopyrightText: Copyright © 2021 Red Hat, Inc., and individual contributors as indicated by the @author tags. + * SPDX-License-Identifier: Apache-2.0 + */ +package org.jboss.pnc.rex.core.counter; + +import jakarta.enterprise.context.ApplicationScoped; +import jakarta.enterprise.inject.Produces; +import lombok.extern.slf4j.Slf4j; +import org.eclipse.microprofile.config.inject.ConfigProperty; +import org.infinispan.counter.api.CounterManager; + +/** + * Selects the active {@link RunningTracker} implementation based on configuration. + *

+ * {@code rex.queue.running-tracker}: + *

    + *
  • {@code cas} (default) - optimistic compare-and-swap on the {@code rex-counter} cache + * (historical behavior, {@link CasRunningTracker});
    • + *
  • {@code strong} - Infinispan clustered StrongCounter ({@link StrongCounterRunningTracker}), + * which removes the single-key contention and scales concurrency.
    • + *
+ * Defaulting to {@code cas} means this can be shipped dark and flipped per environment. + */ +@Slf4j +@ApplicationScoped +public class RunningTrackerProducer { + + public static final String CONFIG_KEY = "rex.queue.running-tracker"; + public static final String STRONG = "strong"; + public static final String CAS = "cas"; + + @Produces + @ApplicationScoped + public RunningTracker runningTracker(@ConfigProperty(name = CONFIG_KEY, defaultValue = CAS) String impl, + @Running Counter casCounter, + CounterManager counterManager) { + if (STRONG.equalsIgnoreCase(impl)) { + log.info("QUEUE: Using StrongCounter running-counter tracker ('{}')", impl); + return new StrongCounterRunningTracker(counterManager); + } + log.info("QUEUE: Using compare-and-swap running-counter tracker ('{}')", impl); + return new CasRunningTracker(casCounter); + } +} diff --git a/core/src/main/java/org/jboss/pnc/rex/core/counter/StrongCounterRunningTracker.java b/core/src/main/java/org/jboss/pnc/rex/core/counter/StrongCounterRunningTracker.java new file mode 100644 index 0000000..f86c274 --- /dev/null +++ b/core/src/main/java/org/jboss/pnc/rex/core/counter/StrongCounterRunningTracker.java @@ -0,0 +1,155 @@ +/* + * SPDX-FileCopyrightText: Copyright © 2021 Red Hat, Inc., and individual contributors as indicated by the @author tags. + * SPDX-License-Identifier: Apache-2.0 + */ +package org.jboss.pnc.rex.core.counter; + +import jakarta.annotation.Nullable; +import lombok.extern.slf4j.Slf4j; +import org.infinispan.counter.api.CounterConfiguration; +import org.infinispan.counter.api.CounterManager; +import org.infinispan.counter.api.CounterType; +import org.infinispan.counter.api.Storage; +import org.infinispan.counter.api.SyncStrongCounter; +import org.jboss.pnc.rex.core.common.Constants; + +import java.util.HashSet; +import java.util.Map; +import java.util.Set; +import java.util.concurrent.ConcurrentHashMap; + +/** + * Infinispan clustered {@link org.infinispan.counter.api.StrongCounter} implementation of + * {@link RunningTracker}. + *

+ * Each queue maps to one unbounded strong counter ({@code RUNNING} for the DEFAULT queue, + * {@code RUNNING-} for named queues). Increments/decrements are atomic cluster-wide, which + * removes the single-key compare-and-swap contention / pessimistic-lock timeouts of the cache-entry + * based implementation. + *

+ * The counters are not transactional, so admission uses reserve-then-compensate + * ({@link #requiresReservation()} is {@code true}) and drift is healed by + * {@code QueueManager#synchronizeRunningCounter()} (run at startup and on demand). + *

+ * Storage is {@link Storage#VOLATILE}: counters are rebuilt from live task state by reconciliation + * after a full-cluster restart. + *

+ * Instantiated by {@link RunningTrackerProducer} (not a CDI bean itself). + */ +@Slf4j +public class StrongCounterRunningTracker implements RunningTracker { + + private static final String DEFAULT_QUEUE_NAMING = "DEFAULT"; + + private static final CounterConfiguration CONFIGURATION = CounterConfiguration + .builder(CounterType.UNBOUNDED_STRONG) + .initialValue(0) + .storage(Storage.VOLATILE) + .build(); + + private final CounterManager counterManager; + + /** Cache of resolved counter name -> blocking counter handle (one per queue per JVM). */ + private final Map counters = new ConcurrentHashMap<>(); + + public StrongCounterRunningTracker(CounterManager counterManager) { + this.counterManager = counterManager; + } + + private String resolveName(String queue) { + if (queue == null) { + return Constants.RUNNING_COUNTER_KEY; + } else if (queue.isBlank()) { + throw new IllegalArgumentException("Counter key must not be blank"); + } + return Constants.RUNNING_COUNTER_KEY + Constants.NAME_SEPARATOR + queue; + } + + private SyncStrongCounter counter(String queue) { + return counters.computeIfAbsent(resolveName(queue), name -> { + // idempotent: returns false if the counter already exists cluster-wide + counterManager.defineCounter(name, CONFIGURATION); + return counterManager.getStrongCounter(name).sync(); + }); + } + + @Override + public long get(@Nullable String queue) { + return counter(queue).getValue(); + } + + @Override + @Nullable + public Long peek(@Nullable String queue) { + String name = resolveName(queue); + if (!counterManager.isDefined(name)) { + return null; + } + return counter(queue).getValue(); + } + + @Override + public long reserve(@Nullable String queue, long n) { + return counter(queue).addAndGet(n); + } + + @Override + public void release(@Nullable String queue, long n) { + addFloored(queue, -n); + } + + @Override + public void decrement(@Nullable String queue) { + addFloored(queue, -1); + } + + /** Add {@code delta} but never let the counter settle below zero (self-heals under-count). */ + private void addFloored(String queue, long delta) { + SyncStrongCounter counter = counter(queue); + long newValue = counter.addAndGet(delta); + if (newValue < 0) { + counter.addAndGet(-newValue); + log.warn("QUEUE '{}': Running counter went below zero ({}); floored to 0.", + queue == null ? DEFAULT_QUEUE_NAMING : queue, newValue); + } + } + + @Override + public void reconcile(@Nullable String queue, long actual) { + SyncStrongCounter counter = counter(queue); + long current = counter.getValue(); + if (current != actual) { + log.info("QUEUE '{}': Synchronizing running counter. Mismatch between active tasks and counter found. " + + "Previous value '{}' -> new value '{}'", + queue == null ? DEFAULT_QUEUE_NAMING : queue, current, actual); + counter.addAndGet(actual - current); + } + } + + @Override + public void initialize(@Nullable String queue, long value) { + reconcile(queue, value); + } + + @Override + public Set queues() { + Set queues = new HashSet<>(); + for (String name : counterManager.getCounterNames()) { + if (!name.startsWith(Constants.RUNNING_COUNTER_KEY)) { + continue; + } + String suffix = name.substring(Constants.RUNNING_COUNTER_KEY.length()); + if (suffix.isEmpty()) { + queues.add(null); // DEFAULT queue + } else if (suffix.startsWith(Constants.NAME_SEPARATOR)) { + queues.add(suffix.substring(Constants.NAME_SEPARATOR.length())); + } + } + return queues; + } + + @Override + public boolean requiresReservation() { + return true; + } +} diff --git a/core/src/main/resources/application.yaml b/core/src/main/resources/application.yaml index 1578d1d..9722ce3 100644 --- a/core/src/main/resources/application.yaml +++ b/core/src/main/resources/application.yaml @@ -37,6 +37,13 @@ scheduler: max-retries: 0 expire-in: 10m # 10minutes +rex: + queue: + # Running-counter implementation: + # cas - optimistic compare-and-swap on the rex-counter cache (default, historical behavior) + # strong - Infinispan clustered StrongCounter (removes single-key contention; scales concurrency) + running-tracker: ${REX_QUEUE_RUNNING_TRACKER:cas} + pnc_client_auth: type: OIDC # or LDAP # ldap_credentials: diff --git a/core/src/test/java/org/jboss/pnc/rex/test/StrongCounterQueueTest.java b/core/src/test/java/org/jboss/pnc/rex/test/StrongCounterQueueTest.java new file mode 100644 index 0000000..b50f97b --- /dev/null +++ b/core/src/test/java/org/jboss/pnc/rex/test/StrongCounterQueueTest.java @@ -0,0 +1,32 @@ +/* + * SPDX-FileCopyrightText: Copyright © 2021 Red Hat, Inc., and individual contributors as indicated by the @author tags. + * SPDX-License-Identifier: Apache-2.0 + */ +package org.jboss.pnc.rex.test; + +import io.quarkus.test.junit.QuarkusTest; +import io.quarkus.test.junit.QuarkusTestProfile; +import io.quarkus.test.junit.TestProfile; + +import java.util.Map; + +/** + * Runs the entire {@link QueueTest} suite with the Infinispan StrongCounter running-counter tracker + * active ({@code rex.queue.running-tracker=strong}), validating the reserve-then-compensate + * admission path and per-queue isolation against a real Infinispan counter subsystem. + *

+ * Test isolation holds because {@code MaintenanceEndpoint} CLEAR_ALL (run before each test) calls + * {@code QueueManager#synchronizeRunningCounter()}, which reconciles every known running counter — + * including StrongCounters — down to the live task count (zero after the caches are cleared). + */ +@QuarkusTest +@TestProfile(StrongCounterQueueTest.StrongTrackerProfile.class) +public class StrongCounterQueueTest extends QueueTest { + + public static class StrongTrackerProfile implements QuarkusTestProfile { + @Override + public Map getConfigOverrides() { + return Map.of("rex.queue.running-tracker", "strong"); + } + } +} diff --git a/core/src/test/java/org/jboss/pnc/rex/test/counter/StrongCounterRunningTrackerTest.java b/core/src/test/java/org/jboss/pnc/rex/test/counter/StrongCounterRunningTrackerTest.java new file mode 100644 index 0000000..2856fac --- /dev/null +++ b/core/src/test/java/org/jboss/pnc/rex/test/counter/StrongCounterRunningTrackerTest.java @@ -0,0 +1,170 @@ +/* + * SPDX-FileCopyrightText: Copyright © 2021 Red Hat, Inc., and individual contributors as indicated by the @author tags. + * SPDX-License-Identifier: Apache-2.0 + */ +package org.jboss.pnc.rex.test.counter; + +import org.infinispan.counter.api.CounterConfiguration; +import org.infinispan.counter.api.CounterManager; +import org.infinispan.counter.api.StrongCounter; +import org.infinispan.counter.api.SyncStrongCounter; +import org.jboss.pnc.rex.core.counter.StrongCounterRunningTracker; +import org.junit.jupiter.api.Test; + +import java.util.List; + +import static org.assertj.core.api.Assertions.assertThat; +import static org.mockito.ArgumentMatchers.any; +import static org.mockito.ArgumentMatchers.anyLong; +import static org.mockito.ArgumentMatchers.anyString; +import static org.mockito.ArgumentMatchers.eq; +import static org.mockito.Mockito.mock; +import static org.mockito.Mockito.never; +import static org.mockito.Mockito.verify; +import static org.mockito.Mockito.when; + +/** + * Pure unit test (no Quarkus/Infinispan server) for {@link StrongCounterRunningTracker}: verifies it + * maps the {@link org.jboss.pnc.rex.core.counter.RunningTracker} operations onto the Infinispan + * clustered StrongCounter API, including the non-negative floor and queue-name parsing. + */ +class StrongCounterRunningTrackerTest { + + private static final String RUNNING = "RUNNING"; + + /** Wires a CounterManager mock whose counters are all backed by the given SyncStrongCounter. */ + private static CounterManager counterManagerReturning(SyncStrongCounter sync) { + CounterManager cm = mock(CounterManager.class); + StrongCounter strong = mock(StrongCounter.class); + when(cm.defineCounter(anyString(), any(CounterConfiguration.class))).thenReturn(true); + when(cm.getStrongCounter(anyString())).thenReturn(strong); + when(strong.sync()).thenReturn(sync); + return cm; + } + + @Test + void getReturnsCounterValue() { + SyncStrongCounter sync = mock(SyncStrongCounter.class); + when(sync.getValue()).thenReturn(3L); + + assertThat(new StrongCounterRunningTracker(counterManagerReturning(sync)).get("q")).isEqualTo(3L); + } + + @Test + void defaultQueueUsesPlainRunningCounterName() { + SyncStrongCounter sync = mock(SyncStrongCounter.class); + CounterManager cm = counterManagerReturning(sync); + + new StrongCounterRunningTracker(cm).get(null); + + verify(cm).getStrongCounter(RUNNING); + } + + @Test + void namedQueueUsesSuffixedCounterName() { + SyncStrongCounter sync = mock(SyncStrongCounter.class); + CounterManager cm = counterManagerReturning(sync); + + new StrongCounterRunningTracker(cm).get("vm"); + + verify(cm).getStrongCounter("RUNNING-vm"); + } + + @Test + void peekReturnsNullWhenCounterNotDefined() { + SyncStrongCounter sync = mock(SyncStrongCounter.class); + CounterManager cm = counterManagerReturning(sync); + when(cm.isDefined("RUNNING-vm")).thenReturn(false); + + assertThat(new StrongCounterRunningTracker(cm).peek("vm")).isNull(); + } + + @Test + void peekReturnsValueWhenCounterDefined() { + SyncStrongCounter sync = mock(SyncStrongCounter.class); + when(sync.getValue()).thenReturn(7L); + CounterManager cm = counterManagerReturning(sync); + when(cm.isDefined("RUNNING-vm")).thenReturn(true); + + assertThat(new StrongCounterRunningTracker(cm).peek("vm")).isEqualTo(7L); + } + + @Test + void reserveAddsAmountAndReturnsNewValue() { + SyncStrongCounter sync = mock(SyncStrongCounter.class); + when(sync.addAndGet(5L)).thenReturn(5L); + + long result = new StrongCounterRunningTracker(counterManagerReturning(sync)).reserve("q", 5); + + assertThat(result).isEqualTo(5L); + verify(sync).addAndGet(5L); + } + + @Test + void releaseSubtractsAmount() { + SyncStrongCounter sync = mock(SyncStrongCounter.class); + when(sync.addAndGet(-3L)).thenReturn(2L); + + new StrongCounterRunningTracker(counterManagerReturning(sync)).release("q", 3); + + verify(sync).addAndGet(-3L); + verify(sync, never()).addAndGet(eq(3L)); // no floor correction needed + } + + @Test + void releaseFloorsAtZeroWhenGoingNegative() { + SyncStrongCounter sync = mock(SyncStrongCounter.class); + when(sync.addAndGet(-3L)).thenReturn(-1L); // underflow + when(sync.addAndGet(1L)).thenReturn(0L); // correction back to zero + + new StrongCounterRunningTracker(counterManagerReturning(sync)).release("q", 3); + + verify(sync).addAndGet(-3L); + verify(sync).addAndGet(1L); + } + + @Test + void decrementSubtractsOne() { + SyncStrongCounter sync = mock(SyncStrongCounter.class); + when(sync.addAndGet(-1L)).thenReturn(4L); + + new StrongCounterRunningTracker(counterManagerReturning(sync)).decrement("q"); + + verify(sync).addAndGet(-1L); + } + + @Test + void reconcileAdjustsByDeltaWhenDifferent() { + SyncStrongCounter sync = mock(SyncStrongCounter.class); + when(sync.getValue()).thenReturn(5L); + + new StrongCounterRunningTracker(counterManagerReturning(sync)).reconcile("q", 2); + + verify(sync).addAndGet(-3L); // 2 - 5 + } + + @Test + void reconcileIsNoopWhenAlreadyEqual() { + SyncStrongCounter sync = mock(SyncStrongCounter.class); + when(sync.getValue()).thenReturn(4L); + + new StrongCounterRunningTracker(counterManagerReturning(sync)).reconcile("q", 4); + + verify(sync, never()).addAndGet(anyLong()); + } + + @Test + void queuesParsesCounterNamesAndMapsDefaultToNull() { + SyncStrongCounter sync = mock(SyncStrongCounter.class); + CounterManager cm = counterManagerReturning(sync); + when(cm.getCounterNames()).thenReturn(List.of(RUNNING, "RUNNING-vm", "SOMETHING_ELSE")); + + assertThat(new StrongCounterRunningTracker(cm).queues()) + .containsExactlyInAnyOrder(null, "vm"); + } + + @Test + void requiresReservationIsTrue() { + assertThat(new StrongCounterRunningTracker(mock(CounterManager.class)).requiresReservation()).isTrue(); + } +} From 51147876119c2b0a7d8425d46211f678ad5d9e00 Mon Sep 17 00:00:00 2001 From: Andrea Vibelli Date: Sun, 11 Oct 2026 11:14:49 +0200 Subject: [PATCH 3/4] Fix CounterManager ambiguous resolution (capacity PoC) Quarkus' Infinispan client already produces a CounterManager CDI bean, so the custom CounterManagerProducer caused an AmbiguousResolutionException at Quarkus augmentation. Remove it and rely on the extension-provided CounterManager (injected directly in RunningTrackerProducer). Verified with `quarkus:build` augmentation locally (the step that failed in CI). Co-Authored-By: Claude Opus 4.8 --- .../core/counter/CounterManagerProducer.java | 26 ------------------- 1 file changed, 26 deletions(-) delete mode 100644 core/src/main/java/org/jboss/pnc/rex/core/counter/CounterManagerProducer.java diff --git a/core/src/main/java/org/jboss/pnc/rex/core/counter/CounterManagerProducer.java b/core/src/main/java/org/jboss/pnc/rex/core/counter/CounterManagerProducer.java deleted file mode 100644 index 46faf08..0000000 --- a/core/src/main/java/org/jboss/pnc/rex/core/counter/CounterManagerProducer.java +++ /dev/null @@ -1,26 +0,0 @@ -/* - * SPDX-FileCopyrightText: Copyright © 2021 Red Hat, Inc., and individual contributors as indicated by the @author tags. - * SPDX-License-Identifier: Apache-2.0 - */ -package org.jboss.pnc.rex.core.counter; - -import jakarta.enterprise.context.ApplicationScoped; -import jakarta.enterprise.inject.Produces; -import org.infinispan.client.hotrod.RemoteCacheManager; -import org.infinispan.client.hotrod.RemoteCounterManagerFactory; -import org.infinispan.counter.api.CounterManager; - -/** - * Exposes the Infinispan {@link CounterManager} (clustered counters) from the Hot Rod - * {@link RemoteCacheManager} that Quarkus produces for the default Infinispan client. The counter - * subsystem lives inside the same Infinispan cluster — no additional deployment is required. - */ -@ApplicationScoped -public class CounterManagerProducer { - - @Produces - @ApplicationScoped - public CounterManager counterManager(RemoteCacheManager cacheManager) { - return RemoteCounterManagerFactory.asCounterManager(cacheManager); - } -} From 9026f0c57abcb336d5b0f0e7f903cdf85e9a454c Mon Sep 17 00:00:00 2001 From: Andrea Vibelli Date: Sun, 11 Oct 2026 11:35:23 +0200 Subject: [PATCH 4/4] Make test HttpEndpoint queue recording tracker-aware (capacity PoC) The mock build-agent endpoint recorded running-counter values by reading the CAS @Running Counter cache directly. Under the StrongCounter profile that cache is empty, so getValue returned null and ConcurrentLinkedQueue.offer(null) threw an NPE inside the mock start handler, making tasks fail to start (START_FAILED) in StrongCounterQueueTest. Read the running count via the active RunningTracker instead (get() returns a primitive, never null), so the queue tests are valid under both the cas and strong running-tracker implementations. Co-Authored-By: Claude Opus 4.8 --- .../org/jboss/pnc/rex/test/endpoints/HttpEndpoint.java | 8 +++----- 1 file changed, 3 insertions(+), 5 deletions(-) diff --git a/core/src/test/java/org/jboss/pnc/rex/test/endpoints/HttpEndpoint.java b/core/src/test/java/org/jboss/pnc/rex/test/endpoints/HttpEndpoint.java index 2aa06b7..9bec4ee 100644 --- a/core/src/test/java/org/jboss/pnc/rex/test/endpoints/HttpEndpoint.java +++ b/core/src/test/java/org/jboss/pnc/rex/test/endpoints/HttpEndpoint.java @@ -14,8 +14,7 @@ import org.jboss.pnc.rex.common.enums.Method; import org.jboss.pnc.rex.common.enums.State; import org.jboss.pnc.rex.core.GenericVertxHttpClient; -import org.jboss.pnc.rex.core.counter.Counter; -import org.jboss.pnc.rex.core.counter.Running; +import org.jboss.pnc.rex.core.counter.RunningTracker; import org.jboss.pnc.rex.model.Header; import org.jboss.pnc.rex.model.requests.RollbackRequest; import org.jboss.pnc.rex.model.requests.StartRequest; @@ -56,8 +55,7 @@ public class HttpEndpoint { ManagedExecutor executor; @Inject - @Running - Counter running; + RunningTracker running; @Inject TransitionRecorder recorder; @@ -222,7 +220,7 @@ private void onResponse(HttpResponse response) { private void record(Object data) { if (shouldRecord) { - record.forEach((key, value) -> value.offer(running.getValue(key))); + record.forEach((key, value) -> value.offer(running.get(key))); recordedRequestData.offer(data); } }