> handlers = new ConcurrentHashMap<>();
+ private final ScheduledExecutorService attemptHeartbeat = Executors.newSingleThreadScheduledExecutor(r -> {
+ final Thread thread = new Thread(r, "command-attempt-heartbeat");
+ thread.setDaemon(true);
+ return thread;
+ });
+
private volatile boolean started = false;
@Override
@@ -75,7 +90,7 @@ public void start() {
return;
}
started = true;
- // run handlers on the shared Micronaut BLOCKING (virtual-thread) executor
+ // run handlers on the shared Micronaut BLOCKING executor
queue.withHandlerExecutor(blockingExecutor);
queue.addConsumer(this::processCommand);
log.info("Command service started - consuming commands");
@@ -97,8 +112,14 @@ public String submit(Command
command) {
final var state = CommandState.submitted(command.id(), command.type(), command.params());
// Persist to storage and submit to queue
- store.save(state);
- queue.submit(CommandMsg.of(command.id(), command.type()));
+ final boolean created = store.create(state);
+ final CommandState existing = created ? state : store.findById(command.id()).orElse(null);
+ // Re-enqueue a pre-existing non-terminal command so a caller retry can repair the
+ // state-created/queue-offer failure window. Duplicate messages are fenced by the
+ // command attempt guard and terminal-state check.
+ if (created || (existing != null && !existing.status().isTerminal())) {
+ queue.submit(CommandMsg.of(command.id(), command.type()));
+ }
log.debug("Command submitted: id={}, type={}", command.id(), command.type());
return command.id();
@@ -119,18 +140,24 @@ public Optional getResult(String commandId, Class resultType) {
@Override
public boolean cancel(String commandId) {
- final var state = store.findById(commandId).orElse(null);
- if (state == null) {
+ final String owner = UUID.randomUUID().toString();
+ if (!store.tryAcquireAttempt(commandId, owner, config.attemptLease())) {
return false;
}
-
- if (state.status().isTerminal()) {
- return false;
+ try {
+ final var state = store.findById(commandId).orElse(null);
+ if (state == null || state.status().isTerminal()) {
+ return false;
+ }
+ if (!store.saveOwned(state.cancelled(), owner)) {
+ return false;
+ }
+ log.info("Command cancelled: id={}", commandId);
+ return true;
+ }
+ finally {
+ store.releaseAttempt(commandId, owner);
}
-
- store.save(state.cancelled());
- log.info("Command cancelled: id={}", commandId);
- return true;
}
@Override
@@ -185,6 +212,37 @@ public P params() {
* @return true to acknowledge (remove from queue), false to retry later
*/
private boolean processCommand(CommandMsg msg) {
+ final String owner = UUID.randomUUID().toString();
+ if (!store.tryAcquireAttempt(msg.commandId(), owner, config.attemptLease())) {
+ log.debug("Command attempt already active, retaining delivery: id={}", msg.commandId());
+ return false;
+ }
+ final ScheduledFuture renewer = attemptHeartbeat.scheduleAtFixedRate(
+ () -> renewAttempt(msg.commandId(), owner),
+ config.attemptHeartbeat().toMillis(),
+ config.attemptHeartbeat().toMillis(),
+ TimeUnit.MILLISECONDS);
+ try {
+ return processCommandOwned(msg, owner);
+ }
+ finally {
+ renewer.cancel(false);
+ store.releaseAttempt(msg.commandId(), owner);
+ }
+ }
+
+ private void renewAttempt(String commandId, String owner) {
+ try {
+ if (!store.renewAttempt(commandId, owner, config.attemptLease())) {
+ log.warn("Command attempt ownership lost while handler is active: id={}", commandId);
+ }
+ }
+ catch (Throwable e) {
+ log.warn("Unable to renew command attempt: id={}; cause={}", commandId, e.getMessage());
+ }
+ }
+
+ private boolean processCommandOwned(CommandMsg msg, String owner) {
// Step 1: Load command state from persistent storage
var state = store.findById(msg.commandId()).orElse(null);
if (state == null) {
@@ -204,14 +262,14 @@ private boolean processCommand(CommandMsg msg) {
final var registration = getHandler(state.type());
if (registration == null) {
log.error("No handler for command type: {}", state.type());
- store.save(state.failed("No handler for type: " + state.type()));
+ saveOwned(state.failed("No handler for type: " + state.type()), owner);
return true;
}
// Step 4: Delegate to the type-capturing helper method
// This pattern allows Java to infer concrete type parameters (P, R) from the
// CommandRegistration, enabling type-safe handler invocation without raw types.
- return processCommandWithHandler(msg, state, registration);
+ return processCommandWithHandler(msg, state, registration, owner);
}
/**
@@ -223,7 +281,8 @@ private boolean processCommand(CommandMsg msg) {
* Runs directly on the shared worker pool thread (no timeout, no extra executor):
*
* - If command is already PROCESSING → call {@code checkStatus()} to poll for completion
- * - If command is still PENDING → call {@code execute()}
+ * - If command is PENDING → persist SUBMITTING, then call {@code execute()}
+ * - If command is SUBMITTING → repeat {@code execute()} with the stable command ID
* - If result status is PROCESSING → mark PROCESSING and return false (re-polled later)
* - If result status is terminal → update state and return true (done)
*
@@ -238,21 +297,26 @@ private boolean processCommand(CommandMsg msg) {
private boolean processCommandWithHandler(
CommandMsg msg,
CommandState state,
- CommandRegistration registration) {
+ CommandRegistration registration,
+ String owner) {
// Reconstruct the typed Command object from persisted state
// Uses Class.cast() internally for type-safe conversion
+ if (state.status() == CommandStatus.PENDING) {
+ state = state.submitting();
+ saveOwned(state, owner);
+ }
final Command command = toCommand(state, registration);
final CommandHandler handler = registration.handler();
try {
- // Branch on the command status. Only PENDING and PROCESSING are reachable here
+ // Branch on the command status. Only SUBMITTING and PROCESSING are reachable here
// (processCommand already acked terminal states); any other value is a bug or a
// newly-added status and must fail loudly rather than be silently executed. Both
// execute() and checkStatus() run on the shared worker pool, so a slow handler
// does not block the loop.
final CommandResult result = switch (state.status()) {
- case PENDING -> handler.execute(command);
+ case SUBMITTING -> handler.execute(command);
case PROCESSING -> handler.checkStatus(command, state);
default -> throw new IllegalStateException("Unexpected command status: " + state.status() + " - id=" + state.id());
};
@@ -262,11 +326,11 @@ private boolean processCommandWithHandler(
// Handler explicitly returned PROCESSING (e.g., async job not yet complete)
// Ensure state reflects PROCESSING status for accurate reporting
if (state.status() != CommandStatus.PROCESSING) {
- store.save(state.started());
+ saveOwned(state.started(), owner);
} else if (state.errorsCount() > 0) {
// Recovered after one or more transient errors — reset the streak. Single write,
// and only when there is something to reset, so healthy re-polls stay write-free.
- store.save(state.clearErrors());
+ saveOwned(state.clearErrors(), owner);
}
return false; // Keep in queue - re-polled and will call checkStatus()
}
@@ -274,7 +338,7 @@ private boolean processCommandWithHandler(
// Terminal result (SUCCEEDED, FAILED, or CANCELLED)
// Apply the result to transition to terminal state
final CommandState newState = state.applyResult(result);
- store.save(newState);
+ saveOwned(newState, owner);
log.debug("Command completed: id={}, status={}", state.id(), newState.status());
return true; // Remove from queue - processing complete
@@ -288,7 +352,7 @@ private boolean processCommandWithHandler(
// non-terminal, stranding the work. Deciding a command has *permanently* failed is
// delegated to the domain layer that owns the entity state (see seqeralabs/sched#712).
log.error("Command processing errored, will retry: id={}", msg.commandId(), e);
- recordError(state, e);
+ recordError(state, e, owner);
return false; // Keep in queue - redelivered / re-polled
}
}
@@ -299,11 +363,17 @@ private boolean processCommandWithHandler(
* command that stays retryable. A failure to persist this must not change control flow: the
* command is kept in the queue and retried regardless.
*/
- private void recordError(CommandState state, Exception e) {
+ private void recordError(CommandState state, Exception e, String owner) {
try {
- store.save(state.withError(e.getMessage() != null ? e.getMessage() : e.toString()));
+ saveOwned(state.withError(e.getMessage() != null ? e.getMessage() : e.toString()), owner);
} catch (Exception fail) {
log.warn("Failed to record command error state: id={}", state.id(), fail);
}
}
+
+ private void saveOwned(CommandState state, String owner) {
+ if (!store.saveOwned(state, owner)) {
+ throw new IllegalStateException("Command attempt ownership lost: id=" + state.id());
+ }
+ }
}
diff --git a/lib-cmd-queue-redis/src/main/java/io/seqera/data/command/CommandState.java b/lib-cmd-queue-redis/src/main/java/io/seqera/data/command/CommandState.java
index f4ca3898..da5d431a 100644
--- a/lib-cmd-queue-redis/src/main/java/io/seqera/data/command/CommandState.java
+++ b/lib-cmd-queue-redis/src/main/java/io/seqera/data/command/CommandState.java
@@ -87,6 +87,13 @@ public CommandState started() {
);
}
+ public CommandState submitting() {
+ return new CommandState(
+ id, type, CommandStatus.SUBMITTING, params,
+ result, error, 0, createdAt, startedAt, Instant.now(), completedAt
+ );
+ }
+
/**
* Transition to SUCCEEDED status with result.
*/
diff --git a/lib-cmd-queue-redis/src/main/java/io/seqera/data/command/CommandStatus.java b/lib-cmd-queue-redis/src/main/java/io/seqera/data/command/CommandStatus.java
index 400c25ab..077002b7 100644
--- a/lib-cmd-queue-redis/src/main/java/io/seqera/data/command/CommandStatus.java
+++ b/lib-cmd-queue-redis/src/main/java/io/seqera/data/command/CommandStatus.java
@@ -31,6 +31,11 @@ public enum CommandStatus {
/** In the queue, awaiting first processing (legacy wire name: {@code "SUBMITTED"}). */
@JsonAlias("SUBMITTED")
PENDING,
+ /**
+ * Initial external submission may have started. Redelivery repeats execute with the
+ * stable command id as idempotency key until PROCESSING is durably recorded.
+ */
+ SUBMITTING,
/** Being processed by a handler (legacy wire name: {@code "RUNNING"}). */
@JsonAlias("RUNNING")
PROCESSING,
diff --git a/lib-cmd-queue-redis/src/main/java/io/seqera/data/command/store/CommandStateStore.java b/lib-cmd-queue-redis/src/main/java/io/seqera/data/command/store/CommandStateStore.java
index 0209c4ac..54f48dcd 100644
--- a/lib-cmd-queue-redis/src/main/java/io/seqera/data/command/store/CommandStateStore.java
+++ b/lib-cmd-queue-redis/src/main/java/io/seqera/data/command/store/CommandStateStore.java
@@ -16,6 +16,7 @@
*/
package io.seqera.data.command.store;
+import java.time.Duration;
import java.util.Optional;
import io.seqera.data.command.CommandState;
@@ -34,6 +35,26 @@ public interface CommandStateStore {
*/
void save(CommandState state);
+ /**
+ * Persist a command only if its id is not already present.
+ */
+ boolean create(CommandState state);
+
+ boolean tryAcquireAttempt(String commandId, String owner, Duration ttl);
+
+ boolean renewAttempt(String commandId, String owner, Duration ttl);
+
+ boolean isAttemptOwner(String commandId, String owner);
+
+ boolean releaseAttempt(String commandId, String owner);
+
+ /**
+ * Persist state only if the supplied attempt still owns the command lease.
+ *
+ * @return true when saved; false when the attempt has lost ownership
+ */
+ boolean saveOwned(CommandState state, String owner);
+
/**
* Find a command state by ID.
*
diff --git a/lib-cmd-queue-redis/src/main/java/io/seqera/data/command/store/CommandStateStoreImpl.java b/lib-cmd-queue-redis/src/main/java/io/seqera/data/command/store/CommandStateStoreImpl.java
index 43c55d79..98f0dd25 100644
--- a/lib-cmd-queue-redis/src/main/java/io/seqera/data/command/store/CommandStateStoreImpl.java
+++ b/lib-cmd-queue-redis/src/main/java/io/seqera/data/command/store/CommandStateStoreImpl.java
@@ -38,11 +38,14 @@
public class CommandStateStoreImpl extends AbstractStateStore implements CommandStateStore {
private static final String PREFIX = "cmd-state/v1";
+ private static final String ATTEMPT_PREFIX = "cmd-attempt/v1:";
private final Duration ttl;
+ private final StateProvider provider;
public CommandStateStoreImpl(StateProvider provider, StringEncodingStrategy encodingStrategy, Duration ttl) {
super(provider, encodingStrategy);
+ this.provider = provider;
this.ttl = ttl;
}
@@ -66,4 +69,39 @@ public void save(CommandState state) {
put(state.id(), state);
}
+ @Override
+ public boolean create(CommandState state) {
+ return putIfAbsent(state.id(), state, ttl);
+ }
+
+ @Override
+ public boolean tryAcquireAttempt(String commandId, String owner, Duration lease) {
+ return provider.putIfAbsent(ATTEMPT_PREFIX + commandId, owner, lease);
+ }
+
+ @Override
+ public boolean renewAttempt(String commandId, String owner, Duration lease) {
+ return provider.compareAndSet(ATTEMPT_PREFIX + commandId, owner, owner, lease);
+ }
+
+ @Override
+ public boolean isAttemptOwner(String commandId, String owner) {
+ return owner.equals(provider.get(ATTEMPT_PREFIX + commandId));
+ }
+
+ @Override
+ public boolean releaseAttempt(String commandId, String owner) {
+ return provider.compareAndDelete(ATTEMPT_PREFIX + commandId, owner);
+ }
+
+ @Override
+ public boolean saveOwned(CommandState state, String owner) {
+ return provider.putIfOwner(
+ ATTEMPT_PREFIX + state.id(),
+ owner,
+ key0(state.id()),
+ serialize(state),
+ ttl);
+ }
+
}
diff --git a/lib-cmd-queue-redis/src/test/groovy/io/seqera/data/command/CommandServiceRedisE2ETest.groovy b/lib-cmd-queue-redis/src/test/groovy/io/seqera/data/command/CommandServiceRedisE2ETest.groovy
new file mode 100644
index 00000000..794bec25
--- /dev/null
+++ b/lib-cmd-queue-redis/src/test/groovy/io/seqera/data/command/CommandServiceRedisE2ETest.groovy
@@ -0,0 +1,93 @@
+/*
+ * Copyright 2026, Seqera Labs
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ */
+package io.seqera.data.command
+
+import java.util.concurrent.CountDownLatch
+import java.util.concurrent.TimeUnit
+import java.util.concurrent.atomic.AtomicInteger
+
+import io.micronaut.context.ApplicationContext
+import io.seqera.fixtures.redis.RedisTestContainer
+import spock.lang.Specification
+import spock.util.concurrent.PollingConditions
+
+class CommandServiceRedisE2ETest extends Specification implements RedisTestContainer {
+
+ def 'should execute a long command once across two replicas and persist terminal state'() {
+ given:
+ def properties = ['redis.uri': "redis://${redisHostName}:${redisPort}".toString()]
+ def ctxA = ApplicationContext.run(properties, 'test', 'redis')
+ def ctxB = ApplicationContext.run(properties, 'test', 'redis')
+ def serviceA = ctxA.getBean(CommandService)
+ def serviceB = ctxB.getBean(CommandService)
+ def calls = new AtomicInteger()
+ def started = new CountDownLatch(1)
+ def release = new CountDownLatch(1)
+ def handler = new RedisOnceHandler(calls, started, release)
+ def id = "redis-e2e-${UUID.randomUUID()}"
+ def command = new TestCommand(id, 'redis-once', new TestParams(1, 'redis'))
+
+ when:
+ serviceA.registerHandler(handler)
+ serviceB.registerHandler(handler)
+ serviceA.start()
+ serviceB.start()
+ serviceA.submit(command)
+ serviceB.submit(command) // duplicate submission is transport-level at-least-once
+
+ then:
+ started.await(5, TimeUnit.SECONDS)
+
+ when: 'the handler remains active for longer than the one-second claim timeout'
+ sleep 2_500
+
+ then: 'neither the Stream lease nor the command-state guard permits overlap'
+ calls.get() == 1
+
+ when:
+ release.countDown()
+
+ then:
+ new PollingConditions(timeout: 8).eventually {
+ assert serviceA.getState(id).orElseThrow().status() == CommandStatus.SUCCEEDED
+ }
+ calls.get() == 1
+
+ cleanup:
+ release.countDown()
+ serviceA?.stop()
+ serviceB?.stop()
+ ctxA?.close()
+ ctxB?.close()
+ }
+}
+
+class RedisOnceHandler implements CommandHandler {
+ private final AtomicInteger calls
+ private final CountDownLatch started
+ private final CountDownLatch release
+
+ RedisOnceHandler(AtomicInteger calls, CountDownLatch started, CountDownLatch release) {
+ this.calls = calls
+ this.started = started
+ this.release = release
+ }
+
+ @Override
+ String type() { 'redis-once' }
+
+ @Override
+ CommandResult execute(Command command) {
+ calls.incrementAndGet()
+ started.countDown()
+ release.await(10, TimeUnit.SECONDS)
+ return CommandResult.success(new TestResult('done', command.params().value))
+ }
+}
diff --git a/lib-cmd-queue-redis/src/test/groovy/io/seqera/data/command/store/CommandAttemptStoreTest.groovy b/lib-cmd-queue-redis/src/test/groovy/io/seqera/data/command/store/CommandAttemptStoreTest.groovy
new file mode 100644
index 00000000..40d3b1e3
--- /dev/null
+++ b/lib-cmd-queue-redis/src/test/groovy/io/seqera/data/command/store/CommandAttemptStoreTest.groovy
@@ -0,0 +1,88 @@
+/*
+ * Copyright 2026, Seqera Labs
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package io.seqera.data.command.store
+
+import java.time.Duration
+
+import io.seqera.data.command.CommandState
+import io.seqera.data.store.state.impl.LocalStateProvider
+import io.seqera.serde.jackson.JacksonEncodingStrategy
+import spock.lang.Specification
+
+class CommandAttemptStoreTest extends Specification {
+
+ private CommandStateStoreImpl newStore() {
+ new CommandStateStoreImpl(
+ new LocalStateProvider(),
+ new JacksonEncodingStrategy() {},
+ Duration.ofHours(1))
+ }
+
+ def 'should create command state idempotently'() {
+ given:
+ def store = newStore()
+ def state = CommandState.submitted('c1', 'test', [value: 1])
+
+ expect:
+ store.create(state)
+ !store.create(state)
+ }
+
+ def 'should fence command attempts by owner and renew their lease'() {
+ given:
+ def store = newStore()
+
+ expect:
+ store.tryAcquireAttempt('c1', 'owner-a', Duration.ofMillis(200))
+ !store.tryAcquireAttempt('c1', 'owner-b', Duration.ofMillis(200))
+ store.isAttemptOwner('c1', 'owner-a')
+
+ and:
+ store.renewAttempt('c1', 'owner-a', Duration.ofSeconds(1))
+ !store.renewAttempt('c1', 'owner-b', Duration.ofSeconds(1))
+ !store.releaseAttempt('c1', 'owner-b')
+ store.releaseAttempt('c1', 'owner-a')
+ store.tryAcquireAttempt('c1', 'owner-b', Duration.ofSeconds(1))
+ }
+
+ def 'should recover an attempt after its owner lease expires'() {
+ given:
+ def store = newStore()
+
+ expect:
+ store.tryAcquireAttempt('c1', 'dead-owner', Duration.ofMillis(50))
+
+ when:
+ sleep 100
+
+ then:
+ store.tryAcquireAttempt('c1', 'live-owner', Duration.ofSeconds(1))
+ }
+
+ def 'should reject a stale owner state update'() {
+ given:
+ def store = newStore()
+ def submitted = CommandState.submitted('c1', 'test', [value: 1])
+ store.create(submitted)
+ store.tryAcquireAttempt('c1', 'owner-a', Duration.ofSeconds(1))
+
+ expect:
+ !store.saveOwned(submitted.failed('stale'), 'owner-b')
+ store.findById('c1').get() == submitted
+ store.saveOwned(submitted.submitting(), 'owner-a')
+ store.findById('c1').get().status().name() == 'SUBMITTING'
+ }
+}
diff --git a/lib-cmd-queue-redis/src/test/java/io/seqera/data/command/TestRedisWorkQueueConfig.java b/lib-cmd-queue-redis/src/test/java/io/seqera/data/command/TestRedisWorkQueueConfig.java
new file mode 100644
index 00000000..9839d245
--- /dev/null
+++ b/lib-cmd-queue-redis/src/test/java/io/seqera/data/command/TestRedisWorkQueueConfig.java
@@ -0,0 +1,35 @@
+/*
+ * Copyright 2026, Seqera Labs
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ */
+package io.seqera.data.command;
+
+import java.time.Duration;
+
+import io.micronaut.context.annotation.Requires;
+import io.seqera.data.workqueue.redis.RedisWorkQueueConfig;
+import jakarta.inject.Singleton;
+
+@Singleton
+@Requires(env = "test")
+public class TestRedisWorkQueueConfig implements RedisWorkQueueConfig {
+ @Override
+ public String getDefaultConsumerGroupName() {
+ return "command-queue-test";
+ }
+
+ @Override
+ public Duration getVisibilityTimeout() {
+ return Duration.ofSeconds(1);
+ }
+
+ @Override
+ public Duration getConsumerWarnTimeout() {
+ return Duration.ofSeconds(5);
+ }
+}
diff --git a/lib-data-store-state-redis/src/main/groovy/io/seqera/data/store/state/impl/LocalStateProvider.groovy b/lib-data-store-state-redis/src/main/groovy/io/seqera/data/store/state/impl/LocalStateProvider.groovy
index 64b96ab4..6b8cd633 100644
--- a/lib-data-store-state-redis/src/main/groovy/io/seqera/data/store/state/impl/LocalStateProvider.groovy
+++ b/lib-data-store-state-redis/src/main/groovy/io/seqera/data/store/state/impl/LocalStateProvider.groovy
@@ -94,6 +94,30 @@ class LocalStateProvider implements StateProvider {
return putIfAbsent0(key, value, ttl) == null
}
+ @Override
+ synchronized boolean compareAndSet(String key, String expected, String value, Duration ttl) {
+ if( get(key) != expected )
+ return false
+ store.put(key, new Entry<>(value, ttl))
+ return true
+ }
+
+ @Override
+ synchronized boolean compareAndDelete(String key, String expected) {
+ if( get(key) != expected )
+ return false
+ store.remove(key)
+ return true
+ }
+
+ @Override
+ synchronized boolean putIfOwner(String ownerKey, String owner, String key, String value, Duration ttl) {
+ if( get(ownerKey) != owner )
+ return false
+ store.put(key, new Entry<>(value, ttl))
+ return true
+ }
+
@Override
synchronized CountResult putJsonIfAbsentAndIncreaseCount(String key, String json, Duration ttl, CountParams counterKey, String luaScript) {
final counter = counterKey.key + '/' + counterKey.field
diff --git a/lib-data-store-state-redis/src/main/groovy/io/seqera/data/store/state/impl/RedisStateProvider.groovy b/lib-data-store-state-redis/src/main/groovy/io/seqera/data/store/state/impl/RedisStateProvider.groovy
index a381035f..87e00a61 100644
--- a/lib-data-store-state-redis/src/main/groovy/io/seqera/data/store/state/impl/RedisStateProvider.groovy
+++ b/lib-data-store-state-redis/src/main/groovy/io/seqera/data/store/state/impl/RedisStateProvider.groovy
@@ -39,6 +39,15 @@ import redis.clients.jedis.params.SetParams
@CompileStatic
class RedisStateProvider implements StateProvider {
+ private static final String COMPARE_SET_SCRIPT =
+ "if redis.call('GET', KEYS[1]) == ARGV[1] then redis.call('SET', KEYS[1], ARGV[2], 'PX', ARGV[3]); return 1 else return 0 end"
+
+ private static final String COMPARE_DELETE_SCRIPT =
+ "if redis.call('GET', KEYS[1]) == ARGV[1] then return redis.call('DEL', KEYS[1]) else return 0 end"
+
+ private static final String PUT_IF_OWNER_SCRIPT =
+ "if redis.call('GET', KEYS[1]) == ARGV[1] then redis.call('SET', KEYS[2], ARGV[2], 'PX', ARGV[3]); return 1 else return 0 end"
+
@Inject
private JedisPool pool
@@ -80,6 +89,30 @@ class RedisStateProvider implements StateProvider {
}
}
+ @Override
+ boolean compareAndSet(String key, String expected, String value, Duration ttl) {
+ try( Jedis conn=pool.getResource() ) {
+ final result = conn.eval(COMPARE_SET_SCRIPT, 1, key, expected, value, ttl.toMillis().toString())
+ return result == 1L
+ }
+ }
+
+ @Override
+ boolean compareAndDelete(String key, String expected) {
+ try( Jedis conn=pool.getResource() ) {
+ final result = conn.eval(COMPARE_DELETE_SCRIPT, 1, key, expected)
+ return result == 1L
+ }
+ }
+
+ @Override
+ boolean putIfOwner(String ownerKey, String owner, String key, String value, Duration ttl) {
+ try( Jedis conn=pool.getResource() ) {
+ final result = conn.eval(PUT_IF_OWNER_SCRIPT, 2, ownerKey, key, owner, value, ttl.toMillis().toString())
+ return result == 1L
+ }
+ }
+
/*
* Set a value only the specified key does not exists, if the value can be set
* the counter identified by the key provided via 'KEYS[2]' is incremented by 1,
diff --git a/lib-data-store-state-redis/src/main/groovy/io/seqera/data/store/state/impl/StateProvider.groovy b/lib-data-store-state-redis/src/main/groovy/io/seqera/data/store/state/impl/StateProvider.groovy
index a528a71d..e09295e7 100644
--- a/lib-data-store-state-redis/src/main/groovy/io/seqera/data/store/state/impl/StateProvider.groovy
+++ b/lib-data-store-state-redis/src/main/groovy/io/seqera/data/store/state/impl/StateProvider.groovy
@@ -29,6 +29,16 @@ import io.seqera.data.store.state.StateStore
*/
interface StateProvider extends StateStore {
+ boolean compareAndSet(K key, V expected, V value, Duration ttl)
+
+ boolean compareAndDelete(K key, V expected)
+
+ /**
+ * Store a value only while an ownership key still contains the expected owner.
+ * The ownership check and value update must be atomic.
+ */
+ boolean putIfOwner(K ownerKey, V owner, K key, V value, Duration ttl)
+
/**
* Store a value in the cache only if does not exist. If the operation is successful
* the counter identified by the key specified is incremented by 1 and the counter (new)
diff --git a/lib-data-store-state-redis/src/test/groovy/io/seqera/data/store/state/impl/LocalStateProviderTest.groovy b/lib-data-store-state-redis/src/test/groovy/io/seqera/data/store/state/impl/LocalStateProviderTest.groovy
index 32ec8257..628e842b 100644
--- a/lib-data-store-state-redis/src/test/groovy/io/seqera/data/store/state/impl/LocalStateProviderTest.groovy
+++ b/lib-data-store-state-redis/src/test/groovy/io/seqera/data/store/state/impl/LocalStateProviderTest.groovy
@@ -179,4 +179,17 @@ class LocalStateProviderTest extends Specification {
provider.get(k) == null
}
+ def 'should update a value only for the current owner'() {
+ given:
+ def ownerKey = UUID.randomUUID().toString()
+ def valueKey = UUID.randomUUID().toString()
+ provider.put(ownerKey, 'owner-a', Duration.ofSeconds(1))
+
+ expect:
+ !provider.putIfOwner(ownerKey, 'owner-b', valueKey, 'stale', Duration.ofSeconds(1))
+ provider.get(valueKey) == null
+ provider.putIfOwner(ownerKey, 'owner-a', valueKey, 'current', Duration.ofSeconds(1))
+ provider.get(valueKey) == 'current'
+ }
+
}
diff --git a/lib-data-store-state-redis/src/test/groovy/io/seqera/data/store/state/impl/RedisStateProviderTest.groovy b/lib-data-store-state-redis/src/test/groovy/io/seqera/data/store/state/impl/RedisStateProviderTest.groovy
index 13ab40d9..410700d3 100644
--- a/lib-data-store-state-redis/src/test/groovy/io/seqera/data/store/state/impl/RedisStateProviderTest.groovy
+++ b/lib-data-store-state-redis/src/test/groovy/io/seqera/data/store/state/impl/RedisStateProviderTest.groovy
@@ -191,5 +191,17 @@ class RedisStateProviderTest extends Specification implements RedisTestContainer
result.count == 3
}
+ def 'should update a value only for the current owner'() {
+ given:
+ def ownerKey = UUID.randomUUID().toString()
+ def valueKey = UUID.randomUUID().toString()
+ provider.put(ownerKey, 'owner-a', Duration.ofSeconds(1))
+
+ expect:
+ !provider.putIfOwner(ownerKey, 'owner-b', valueKey, 'stale', Duration.ofSeconds(1))
+ provider.get(valueKey) == null
+ provider.putIfOwner(ownerKey, 'owner-a', valueKey, 'current', Duration.ofSeconds(1))
+ provider.get(valueKey) == 'current'
+ }
}
diff --git a/lib-data-workqueue-redis/src/main/java/io/seqera/data/workqueue/redis/RedisWorkQueue.java b/lib-data-workqueue-redis/src/main/java/io/seqera/data/workqueue/redis/RedisWorkQueue.java
index ba56d62c..7f79be18 100644
--- a/lib-data-workqueue-redis/src/main/java/io/seqera/data/workqueue/redis/RedisWorkQueue.java
+++ b/lib-data-workqueue-redis/src/main/java/io/seqera/data/workqueue/redis/RedisWorkQueue.java
@@ -18,6 +18,7 @@
package io.seqera.data.workqueue.redis;
import java.time.Duration;
+import java.util.Collections;
import java.util.List;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;
@@ -37,7 +38,6 @@
import redis.clients.jedis.StreamEntryID;
import redis.clients.jedis.exceptions.JedisDataException;
import redis.clients.jedis.params.XAutoClaimParams;
-import redis.clients.jedis.params.XClaimParams;
import redis.clients.jedis.params.XReadGroupParams;
import redis.clients.jedis.resps.StreamEntry;
@@ -86,6 +86,17 @@ public class RedisWorkQueue implements WorkQueue {
private static final String DATA_FIELD = "data";
+ /**
+ * Touch a pending entry only if this consumer still owns it. Without the ownership
+ * check, a stale handler could XCLAIM the entry back after a peer reclaimed it.
+ * The script uses Stream/PEL commands only.
+ */
+ private static final String RENEW_IF_OWNER_SCRIPT =
+ "local p=redis.call('XPENDING',KEYS[1],ARGV[1],ARGV[2],ARGV[2],1); " +
+ "if #p==1 and p[1][2]==ARGV[3] then " +
+ "return redis.call('XCLAIM',KEYS[1],ARGV[1],ARGV[3],0,ARGV[2],'JUSTID') " +
+ "else return {} end";
+
@Inject
private JedisPool pool;
@@ -149,12 +160,20 @@ public void offer(String queueId, String message) {
* The returned lease id is the Redis {@link StreamEntryID} of the delivered entry.
*/
@Override
- public Lease receive(String queueId) {
+ public Lease receiveNew(String queueId) {
try (Jedis jedis = pool.getResource()) {
- StreamEntry entry = claimMessage(jedis, queueId);
+ final StreamEntry entry = readMessage(jedis, queueId);
if (entry == null) {
- entry = readMessage(jedis, queueId);
+ return null;
}
+ return new Lease<>(entry.getID().toString(), entry.getFields().get(DATA_FIELD));
+ }
+ }
+
+ @Override
+ public Lease reclaim(String queueId) {
+ try (Jedis jedis = pool.getResource()) {
+ final StreamEntry entry = claimMessage(jedis, queueId);
if (entry == null) {
return null;
}
@@ -173,13 +192,10 @@ public Lease receive(String queueId) {
@Override
public void renewLease(String queueId, String leaseId) {
try (Jedis jedis = pool.getResource()) {
- jedis.xclaimJustId(
- queueId,
- config.getDefaultConsumerGroupName(),
- consumerName,
- 0L,
- XClaimParams.xClaimParams(),
- new StreamEntryID(leaseId));
+ jedis.eval(
+ RENEW_IF_OWNER_SCRIPT,
+ Collections.singletonList(queueId),
+ List.of(config.getDefaultConsumerGroupName(), leaseId, consumerName));
}
}
@@ -205,12 +221,13 @@ public void ack(String queueId, String leaseId) {
/**
* {@inheritDoc}
*
- * No-op: the entry remains in the pending-entries list and becomes reclaimable
- * by a peer consumer once its idle time exceeds the visibility timeout.
+ *
Touches the entry one final time and leaves it in the pending-entries list.
+ * It becomes reclaimable after a full visibility timeout without retaining any
+ * Java-side lifecycle state.
*/
@Override
public void release(String queueId, String leaseId) {
- // no-op: entry stays in the PEL, reclaimable after the visibility timeout
+ renewLease(queueId, leaseId);
}
/**
diff --git a/lib-data-workqueue-redis/src/main/java/io/seqera/data/workqueue/redis/RedisWorkQueueConfig.java b/lib-data-workqueue-redis/src/main/java/io/seqera/data/workqueue/redis/RedisWorkQueueConfig.java
index 57207a2e..18088572 100644
--- a/lib-data-workqueue-redis/src/main/java/io/seqera/data/workqueue/redis/RedisWorkQueueConfig.java
+++ b/lib-data-workqueue-redis/src/main/java/io/seqera/data/workqueue/redis/RedisWorkQueueConfig.java
@@ -119,10 +119,9 @@ default long getHeartbeatIntervalMillis() {
}
/**
- * Returns the upper bound on a single {@code accept()} invocation before its
- * lease is released (safety valve). This bounds one handler invocation, not the
- * total lease lifetime; past this bound the heartbeat daemon stops renewing the
- * lease so it becomes reclaimable. Defaults to {@code 15m}.
+ * Returns the duration after which a still-running {@code accept()} invocation is
+ * reported as stalled. Renewal continues while the handler is active to prevent
+ * timeout-driven overlap. Defaults to {@code 15m}.
*
* @return the maximum single-invocation processing time duration
*/
diff --git a/lib-data-workqueue-redis/src/test/groovy/io/seqera/data/workqueue/AsyncWorkQueueRedisTest.groovy b/lib-data-workqueue-redis/src/test/groovy/io/seqera/data/workqueue/AsyncWorkQueueRedisTest.groovy
index 7c28ab06..8325d73c 100644
--- a/lib-data-workqueue-redis/src/test/groovy/io/seqera/data/workqueue/AsyncWorkQueueRedisTest.groovy
+++ b/lib-data-workqueue-redis/src/test/groovy/io/seqera/data/workqueue/AsyncWorkQueueRedisTest.groovy
@@ -19,6 +19,7 @@ package io.seqera.data.workqueue
import java.time.Duration
import java.util.concurrent.CountDownLatch
+import java.util.concurrent.ConcurrentHashMap
import java.util.concurrent.TimeUnit
import java.util.concurrent.atomic.AtomicInteger
@@ -47,6 +48,48 @@ class AsyncWorkQueueRedisTest extends Specification implements RedisTestContaine
return ApplicationContext.run('test', 'redis')
}
+ def 'should admit all live messages while concurrency only bounds active invocations'() {
+ given:
+ def ctx = newContext()
+ def target = ctx.getBean(RedisWorkQueue)
+ def queue = new TunableQueue(target,
+ concurrency: 2,
+ pollInterval: Duration.ofMillis(50),
+ heartbeatInterval: Duration.ofMillis(300))
+ def id = "queue-${LongRndKey.rndHex()}"
+ def firstSeen = ConcurrentHashMap.newKeySet()
+ def admitted = new CountDownLatch(20)
+ def active = new AtomicInteger()
+ def maxActive = new AtomicInteger()
+
+ when:
+ queue.addConsumer(id, { msg ->
+ def n = active.incrementAndGet()
+ maxActive.accumulateAndGet(n, Math::max)
+ try {
+ if (firstSeen.add(msg)) {
+ admitted.countDown()
+ }
+ return false
+ }
+ finally {
+ active.decrementAndGet()
+ }
+ })
+ 20.times { queue.offer(id, "task-$it".toString()) }
+
+ then: 'every task gets an invocation even though all remain non-terminal'
+ admitted.await(8, TimeUnit.SECONDS)
+ firstSeen.size() == 20
+
+ and: 'only handler calls, not live tasks, are bounded'
+ maxActive.get() <= 2
+
+ cleanup:
+ queue.close()
+ ctx.stop()
+ }
+
// single instance — a handler running longer than visibility-timeout is not
// reclaimed by this instance's own poll while it is alive/heartbeated
def 'should not reclaim live work within a single instance' () {
@@ -165,9 +208,8 @@ class AsyncWorkQueueRedisTest extends Specification implements RedisTestContaine
ctxLive.stop()
}
- // a single invocation exceeding max-processing-time has its lease released
- // (stops being renewed), so the message is reclaimed and re-delivered
- def 'should release the lease of an invocation exceeding max-processing-time' () {
+ // max-processing-time is diagnostic only: releasing a live handler would overlap it.
+ def 'should continue renewing beyond max-processing-time without overlapping execution' () {
given:
def ctx = newContext()
def target = ctx.getBean(RedisWorkQueue)
@@ -179,25 +221,28 @@ class AsyncWorkQueueRedisTest extends Specification implements RedisTestContaine
def id = "queue-${LongRndKey.rndHex()}"
def calls = new AtomicInteger()
def hang = new CountDownLatch(1)
- def redelivered = new CountDownLatch(1)
+ def completed = new CountDownLatch(1)
when:
queue.addConsumer(id, { msg ->
- def n = calls.incrementAndGet()
- if (n == 1) {
- // first invocation hangs past max-processing-time (1s) -> lease released
- hang.await()
- return true
- }
- // the re-delivered invocation completes normally
- redelivered.countDown()
+ calls.incrementAndGet()
+ hang.await()
+ completed.countDown()
return true
})
queue.offer(id, 'hung')
- then: 'the stalled invocation is evicted and the message is re-delivered'
- redelivered.await(10, TimeUnit.SECONDS)
- calls.get() >= 2
+ then: 'the live invocation remains the sole owner well beyond both timeouts'
+ sleep 3_000
+ calls.get() == 1
+
+ when:
+ hang.countDown()
+
+ then:
+ completed.await(5, TimeUnit.SECONDS)
+ sleep 1_500
+ calls.get() == 1
cleanup:
hang.countDown()
diff --git a/lib-data-workqueue-redis/src/test/groovy/io/seqera/data/workqueue/TunableQueue.groovy b/lib-data-workqueue-redis/src/test/groovy/io/seqera/data/workqueue/TunableQueue.groovy
index 4e2523a9..431fbaf7 100644
--- a/lib-data-workqueue-redis/src/test/groovy/io/seqera/data/workqueue/TunableQueue.groovy
+++ b/lib-data-workqueue-redis/src/test/groovy/io/seqera/data/workqueue/TunableQueue.groovy
@@ -24,7 +24,7 @@ import io.seqera.serde.encode.StringEncodingStrategy
/**
* A {@link AbstractWorkQueue} used by the async-processing tests. It carries a
* String payload (identity encoding) and exposes the async knobs — concurrency,
- * poll interval, heartbeat interval and max-processing-time — as constructor options
+ * poll interval, heartbeat interval and handler-warning threshold — as constructor options
* so each test can tune them independently.
*
* @author Paolo Di Tommaso
diff --git a/lib-data-workqueue/README.md b/lib-data-workqueue/README.md
index 8a1abc7f..58886304 100644
--- a/lib-data-workqueue/README.md
+++ b/lib-data-workqueue/README.md
@@ -56,7 +56,7 @@ stream key, e.g. `cmd-queue/v1`).
|---|---|---|---|---|
| `seqera.workqueue.entries` | Gauge | — | entries | Current queue backlog (Redis `XLEN`, polled at scrape time). |
| `seqera.workqueue.messages` | Counter | `outcome` | messages | Total messages processed per outcome. |
-| `seqera.workqueue.processing` | Timer | `outcome` | seconds | Per-entry processing time. Includes the full lifecycle from the underlying `queue.consume(...)` entry through the consumer's `accept` and the Redis acknowledge/delete. Published as a Prometheus histogram (with buckets) so quantiles can be aggregated server-side across replicas via `histogram_quantile()`. |
+| `seqera.workqueue.processing` | Timer | `outcome` | seconds | Per-handler-invocation processing time. Published as a Prometheus histogram (with buckets) so quantiles can be aggregated server-side across replicas via `histogram_quantile()`. |
The `outcome` tag takes one of three values:
@@ -154,11 +154,9 @@ workQueue.addConsumer("user-activity", new ActivityConsumer())
## Architecture
-`AbstractWorkQueue` runs handlers **asynchronously and concurrently** while
-guaranteeing that a given message is processed by exactly one *live* consumer at a
-time. A message is owned by its consumer for as long as the handler keeps working —
-independent of how long that takes — and ownership is relinquished only when the work
-finishes or the consumer dies.
+`AbstractWorkQueue` is an at-least-once transport. `concurrency()` bounds active
+handler invocations, not the number of live messages. A non-terminal message consumes
+no Java thread or semaphore permit between invocations.
```
offer(msg) ┌──────────────────────────────┐
@@ -171,11 +169,11 @@ finishes or the consumer dies.
│ group) │◀──────────── heartbeat daemon ────┤ (never runs it inline) │
│ │ every visibility-timeout/3 │ │
│ │ ack (XACK + XDEL) │ worker (executor thread) │
- │ │◀──────────── on terminal ─────────┤ accept(msg): │
- └──────────┘ │ ├─ true → ack + free slot │
- ▲ │ └─ false → keep lease, │
- │ reclaimed by a peer only if the owner │ re-run after pollInterval│
- │ dies (heartbeat stops → idle > visibility-timeout) via the re-poll sched │
+ │ │◀──────────── on terminal ─────────┤ one accept(msg) call: │
+ └──────────┘ │ ├─ true → ack │
+ ▲ │ └─ false → leave in PEL │
+ │ retry/dead-owner reclaim after │ then always free the slot │
+ │ idle > visibility-timeout │ │
└─────────────────────────────────────────────────────────────────────────┘
```
@@ -185,27 +183,23 @@ finishes or the consumer dies.
handler; it hands each message to a worker executor and moves on. Handlers run on the
executor supplied via `withHandlerExecutor(...)` — **mandatory, no default** (Micronaut
consumers inject the `@Named(BLOCKING)` executor). A `Semaphore` sized by
- `concurrency()` bounds how many messages are in flight at once (backpressure: excess
- messages stay in the queue).
+ `concurrency()` bounds how many handler calls run at once.
2. **Heartbeat lease (single live runner + safe long handlers).** While a message is in
flight, a daemon renews its Redis consumer-group entry (`XCLAIM … JUSTID`) every
`visibility-timeout / 3`, pinning its idle time near zero so no peer's `XAUTOCLAIM` can
reclaim it — no matter how long the handler runs. If the owning process dies, the
heartbeat stops, idle time crosses the visibility timeout, and a peer reclaims the message
- (real dead-consumer failover). A `max-processing-time` safety valve stops renewing a
- single invocation that runs pathologically long (logged as *stalled*), without
- interrupting its thread.
-
-3. **In-process re-poll for not-yet-terminal work.** When a handler returns `false` (work
- in progress), the message keeps its lease and the handler is **re-invoked in-process**
- after `pollInterval` via a scheduler — Redis is not re-read. This makes the re-poll
- cadence independent of `visibility-timeout` (which then governs only failover). The next
- invocation is scheduled only after the previous one returns, so a given message is
- never processed by two overlapping invocations.
-
-Delivery is **at-least-once** (a crash/pause beyond `visibility-timeout`, or the
-`max-processing-time` valve, can hand a still-running message to a peer), so consumers
+ (real dead-consumer failover). `max-processing-time` is a warning threshold only:
+ renewal continues while the handler thread is active, preventing timeout-driven overlap.
+
+3. **Stream-owned retry.** When a handler returns `false`, the entry is touched once and
+ left in the Redis PEL. Its thread and permit are released immediately. A later
+ `XAUTOCLAIM` starts the next invocation after `visibility-timeout`. New delivery and
+ expired reclaim are selected with a weighted policy so retries cannot freeze intake.
+
+Delivery is **at-least-once** (a crash, partition, or pause beyond
+`visibility-timeout` can hand a still-running message to a peer), so consumers
must be idempotent. The in-memory `LocalWorkQueue` has no pending-entries list, so it
has no lease/heartbeat (renewLease is a no-op); it still benefits from async, concurrent dispatch.
@@ -213,11 +207,12 @@ has no lease/heartbeat (renewLease is a no-op); it still benefits from async, co
| Knob | Where | Default | Governs |
|---|---|---|---|
-| `pollInterval()` | `AbstractWorkQueue` | — (subclass) | Idle backoff **and** in-process re-poll cadence |
-| `concurrency()` | `AbstractWorkQueue` | `1` | Max in-flight messages (semaphore ceiling) |
-| `getVisibilityTimeout()` | `RedisWorkQueueConfig` | — | Dead-consumer failover window |
+| `pollInterval()` | `AbstractWorkQueue` | — (subclass) | Idle dispatcher backoff |
+| `concurrency()` | `AbstractWorkQueue` | `1` | Max concurrent handler invocations |
+| `newToRetryRatio()` | `AbstractWorkQueue` | `3` | Fair selection between new and expired entries |
+| `getVisibilityTimeout()` | `RedisWorkQueueConfig` | — | Retry cadence and dead-consumer failover |
| `getHeartbeatInterval()` | `RedisWorkQueueConfig` | `visibility-timeout / 3` | Lease renewal cadence |
-| `getMaxProcessingTime()` | `RedisWorkQueueConfig` | `15m` | Upper bound on a single `accept()` before its lease is released |
+| `getMaxProcessingTime()` | `RedisWorkQueueConfig` | `15m` | Long-handler warning threshold |
## Testing
diff --git a/lib-data-workqueue/changelog.txt b/lib-data-workqueue/changelog.txt
index b7755cf5..0920bb3f 100644
--- a/lib-data-workqueue/changelog.txt
+++ b/lib-data-workqueue/changelog.txt
@@ -1,5 +1,13 @@
# lib-data-workqueue changelog
+Next
+- Concurrency now bounds active handler invocations rather than live message lifecycles.
+- Non-terminal entries are returned to the backend after each invocation; Redis entries
+ remain in the PEL and are reclaimed after visibility timeout.
+- Heartbeats exist only while a handler runs, allowing execution beyond visibility timeout.
+- max-processing-time is diagnostic and never releases a still-running handler.
+- Split new delivery from expired reclaim and select them fairly so retries cannot starve intake.
+
1.0.0 - 11 Jul 2026
- Initial release. This module is the redis-free split of lib-data-stream-redis 1.6.0,
renamed to reflect the reliable work-queue semantics it actually implements (competing
diff --git a/lib-data-workqueue/src/main/java/io/seqera/data/workqueue/AbstractWorkQueue.java b/lib-data-workqueue/src/main/java/io/seqera/data/workqueue/AbstractWorkQueue.java
index 760f0fc3..42662ced 100644
--- a/lib-data-workqueue/src/main/java/io/seqera/data/workqueue/AbstractWorkQueue.java
+++ b/lib-data-workqueue/src/main/java/io/seqera/data/workqueue/AbstractWorkQueue.java
@@ -127,10 +127,8 @@ public abstract class AbstractWorkQueue implements Closeable {
private final String name0;
/**
- * A message picked up from a queue and held while it is processed. The
- * {@code queueId} + {@code leaseId} pair identifies the delivered entry; the
- * {@code message} is kept so a not-yet-terminal command can be re-invoked in-process
- * (Model B) without re-reading it from the queue.
+ * One active handler invocation. It exists only while {@code accept()} is running,
+ * never for the full lifecycle of a non-terminal message.
*/
private record InFlight(String queueId, String leaseId, String message) {
String key() {
@@ -139,17 +137,14 @@ String key() {
}
/**
- * Leases held from pickup to terminal/crash; every entry is heartbeated by the
- * daemon so an alive consumer is never reclaimed. Keyed by {@code queueId|leaseId}.
+ * Handler invocations currently running. Every entry is heartbeated until that one
+ * invocation returns, allowing handler time to exceed the queue visibility timeout.
*/
- private final Map inFlight = new ConcurrentHashMap<>();
+ private final Map active = new ConcurrentHashMap<>();
- /**
- * Subset of {@link #inFlight} whose {@code accept()} invocation is running right now,
- * mapped to the wall-clock millis at which that invocation started. Used by the
- * heartbeat daemon to enforce {@code max-processing-time} on a single invocation.
- */
- private final Map active = new ConcurrentHashMap<>();
+ private final Map activeSince = new ConcurrentHashMap<>();
+
+ private final java.util.Set warned = ConcurrentHashMap.newKeySet();
/**
* Executor that runs the message handlers. Supplied by the consumer via
@@ -161,25 +156,21 @@ String key() {
private volatile ExecutorService pool;
/**
- * Gates new intake: a permit is acquired when a lease is picked up and held for the
- * whole lease lifetime (across re-polls), released on terminal ack / eviction /
- * release. This bounds concurrent handlers to {@link #concurrency()} and reserves
- * capacity so in-flight commands' re-polls are never starved by new intake.
+ * Bounds handler invocations, not live messages. A permit is held only from delivery
+ * until one {@code accept()} call returns.
*/
private volatile Semaphore slots;
/**
- * Schedules delayed re-poll re-submissions for not-yet-terminal commands (Model B).
- */
- private volatile ScheduledExecutorService scheduler;
-
- /**
- * Renews every in-flight lease on a fixed cadence so an alive consumer keeps ownership.
+ * Renews every active invocation on a fixed cadence so a handler may safely run longer
+ * than the visibility timeout.
*/
private volatile ScheduledExecutorService heartbeat;
private volatile boolean closed;
+ private final AtomicInteger deliverySequence = new AtomicInteger();
+
/**
* Constructs a new queue without metrics instrumentation. Behavior is identical
* to passing {@link NoopQueueMetrics#INSTANCE} to {@link #AbstractWorkQueue(WorkQueue, QueueMetrics)}.
@@ -225,21 +216,28 @@ protected Thread createListenerThread() {
/**
* @return
* The time interval to await before trying to read again the queue
- * when no more entries are available. Also the cadence at which a
- * not-yet-terminal command is re-invoked in-process (Model B).
+ * when no more entries are available. Redis retry cadence is governed
+ * by the visibility timeout of the pending Stream entry.
*/
protected abstract Duration pollInterval();
/**
* @return
- * The maximum number of message handlers that may run concurrently on this
- * instance (the worker pool size). Defaults to {@code 1}; subclasses may
- * override to enable parallel processing.
+ * The maximum number of handler invocations that may run concurrently on this
+ * instance. A non-terminal message releases its permit after each invocation.
*/
protected int concurrency() {
return 1;
}
+ /**
+ * Number of new-message delivery opportunities for each expired-message reclaim.
+ * Both paths fall back to the other when empty, so neither intake nor retry can starve.
+ */
+ protected int newToRetryRatio() {
+ return 3;
+ }
+
/**
* @return
* How often in-flight leases are renewed so an alive consumer keeps ownership
@@ -254,9 +252,9 @@ protected Duration heartbeatInterval() {
/**
* @return
- * The upper bound on a single {@code accept()} invocation before its lease is
- * released (safety valve); it does not interrupt the handler thread. Defaults
- * to {@code 15m}.
+ * Duration after which a still-running invocation is reported as stalled.
+ * This is an observability threshold only: the lease continues to be renewed,
+ * because releasing it while the handler runs would create overlapping execution.
*/
protected Duration maxProcessingTime() {
final Duration d = queue.maxProcessingTime();
@@ -325,17 +323,15 @@ public void addConsumer(String queueId, MessageConsumer consumer) {
}
/**
- * Lazily create the worker pool, the re-poll scheduler, the heartbeat daemon and the
- * capacity gate, then start the dispatcher thread. Invoked once, when the first
+ * Lazily create the heartbeat daemon and invocation-capacity gate, then start the
+ * dispatcher thread. Invoked once, when the first
* consumer is registered.
*/
private void startProcessing() {
// a handler executor must be supplied via withHandlerExecutor() before processing starts
Objects.requireNonNull(pool, "Handler executor not set - call withHandlerExecutor() before addConsumer()");
- // 'slots' — not the executor — bounds how many commands may be in flight at once;
- // the cap is a memory/heartbeat ceiling, independent of the executor's threading model.
+ // slots bound active handler calls, never the number of live queue entries
this.slots = new Semaphore(Math.max(1, concurrency()));
- this.scheduler = new ScheduledThreadPoolExecutor(1, daemonFactory(name() + "-repoll-" + count.get()));
this.heartbeat = new ScheduledThreadPoolExecutor(1, daemonFactory(name() + "-heartbeat-" + count.get()));
final long hb = heartbeatInterval().toMillis();
this.heartbeat.scheduleAtFixedRate(this::heartbeatTick, hb, hb, TimeUnit.MILLISECONDS);
@@ -383,9 +379,8 @@ protected boolean processMessage(String msg, MessageConsumer consumer) {
/**
* The dispatcher loop (runs on the listener thread). It never runs a handler itself:
- * for every queue that has free pool capacity it polls one message (without acking)
- * and submits its processing to the worker pool, then sleeps for {@link #pollInterval()}
- * when nothing was polled this cycle.
+ * for every queue that has free invocation capacity it selects fairly between new
+ * messages and expired pending messages, then submits one handler call to the pool.
*/
protected void processMessages() {
log.trace("Work queue - starting dispatcher thread");
@@ -393,9 +388,7 @@ protected void processMessages() {
try {
boolean polled = false;
for (Map.Entry> entry : listeners.entrySet()) {
- // poll a queue only when a worker slot is free (backpressure); the
- // permit is held for the whole lease lifetime so re-polls of in-flight
- // commands are never starved by new intake
+ // A permit is held for one handler invocation only.
if (!slots.tryAcquire()) {
break;
}
@@ -425,8 +418,8 @@ protected void processMessages() {
}
/**
- * Poll a single queue (a worker permit has already been acquired by the caller) and,
- * if a message is available, register it as in-flight and submit it to the pool.
+ * Poll a single queue (an invocation permit has already been acquired by the caller)
+ * and submit one delivery to the pool.
* If nothing is available the permit is released and {@code false} is returned.
*
* @return {@code true} if a message was polled and submitted, {@code false} otherwise
@@ -434,21 +427,22 @@ protected void processMessages() {
private boolean dispatchOne(String queueId) {
boolean submitted = false;
try {
- final WorkQueue.Lease lease = queue.receive(queueId);
+ final WorkQueue.Lease lease = receiveFair(queueId);
if (lease == null) {
metrics.recordOutcome(metrics.startSample(), queueId, Outcome.EMPTY);
return false;
}
final var e = new InFlight(queueId, lease.id(), lease.message());
- // Guard against self-reclaim: if the heartbeat falls behind by more than the
- // visibility timeout, this instance's own receive() (XAUTOCLAIM) can re-deliver an
- // entry it is already processing. The reclaim only refreshed the lease idle time, so
- // keep the live in-flight entry and drop the duplicate — otherwise a second handler
- // runs concurrently and its permit leaks (the original remove() returns null).
- if (inFlight.putIfAbsent(e.key(), e) != null) {
+ // Guard against a local self-reclaim after a delayed heartbeat.
+ if (active.putIfAbsent(e.key(), e) != null) {
return false; // 'submitted' stays false → finally releases this permit
}
- submitRun(e);
+ activeSince.put(e.key(), System.currentTimeMillis());
+ if (!submitRun(e)) {
+ activeSince.remove(e.key());
+ active.remove(e.key());
+ return false;
+ }
submitted = true;
return true;
}
@@ -461,33 +455,52 @@ private boolean dispatchOne(String queueId) {
}
}
+ private WorkQueue.Lease receiveFair(String queueId) {
+ final int ratio = Math.max(1, newToRetryRatio());
+ final boolean preferNew = Math.floorMod(deliverySequence.getAndIncrement(), ratio + 1) < ratio;
+ WorkQueue.Lease result = preferNew
+ ? queue.receiveNew(queueId)
+ : queue.reclaim(queueId);
+ if (result == null) {
+ result = preferNew
+ ? queue.reclaim(queueId)
+ : queue.receiveNew(queueId);
+ }
+ return result;
+ }
+
/**
* Submit the processing of an in-flight lease to the worker pool. Swallows the
* rejection that occurs when the pool is being shut down.
*/
- private void submitRun(InFlight e) {
+ private boolean submitRun(InFlight e) {
try {
pool.execute(() -> run(e));
+ return true;
}
catch (RejectedExecutionException ex) {
log.debug("Work queue - worker pool rejected task for entry={} (shutting down)", e.key());
+ return false;
}
}
/**
- * Runs a single {@code accept()} invocation on a worker thread. On {@code true}
- * (terminal) it acks the message and drops the lease; on {@code false} (Model B,
- * not-yet-terminal) it keeps the lease in-flight and schedules the next invocation
- * after {@link #pollInterval()} — strictly serial per command, since the next
- * invocation is scheduled only after this one returned.
+ * Runs exactly one {@code accept()} invocation. A terminal result is acknowledged;
+ * a non-terminal result is touched once and left in the PEL for reclaim after the
+ * visibility timeout. Either outcome releases the invocation permit.
*/
private void run(InFlight e) {
- final boolean accepted = invokeHandler(e);
- if (accepted) {
- acknowledge(e);
+ try {
+ final boolean accepted = invokeHandler(e);
+ if (accepted) {
+ acknowledge(e);
+ }
+ else if (!closed) {
+ retryLater(e);
+ }
}
- else if (shouldRepoll(e)) {
- scheduleRepoll(e);
+ finally {
+ finishAttempt(e);
}
}
@@ -501,7 +514,6 @@ private boolean invokeHandler(InFlight e) {
final long sample = metrics.startSample();
boolean accepted = false;
Outcome outcome = Outcome.ACTIVE;
- active.put(e.key(), System.currentTimeMillis());
try {
accepted = processMessage(e.message(), consumer);
outcome = accepted ? Outcome.PROCESSED : Outcome.ACTIVE;
@@ -511,7 +523,6 @@ private boolean invokeHandler(InFlight e) {
log.error("Work queue - error processing entry={} - cause: {}", e.key(), t.getMessage(), t);
}
finally {
- active.remove(e.key());
metrics.recordOutcome(sample, e.queueId(), outcome);
}
return accepted;
@@ -525,78 +536,62 @@ private void acknowledge(InFlight e) {
catch (Throwable t) {
log.error("Work queue - error acking entry={} - cause: {}", e.key(), t.getMessage(), t);
}
- finally {
- releaseLease(e.key());
- }
- }
-
- /** Whether a not-yet-terminal command should be re-polled: still owned and not shutting down. */
- private boolean shouldRepoll(InFlight e) {
- return !closed && inFlight.containsKey(e.key());
}
/**
- * Keep the lease (the heartbeat keeps renewing it, so no reclaim/migration) and schedule
- * the next in-process invocation after {@link #pollInterval()} — strictly serial, since
- * it is scheduled only after the previous invocation returned.
+ * Reset the delivery idle time after a non-terminal result, then leave the entry in
+ * the Stream PEL. It becomes eligible for another invocation after visibility timeout.
*/
- private void scheduleRepoll(InFlight e) {
+ private void retryLater(InFlight e) {
try {
- scheduler.schedule(() -> submitRun(e), pollInterval().toMillis(), TimeUnit.MILLISECONDS);
+ queue.release(e.queueId(), e.leaseId());
}
- catch (RejectedExecutionException ex) {
- log.debug("Work queue - re-poll scheduler rejected entry={} (shutting down)", e.key());
+ catch (Throwable t) {
+ log.warn("Work queue - error touching retry entry={} - cause: {}", e.key(), t.getMessage());
}
}
- /**
- * Drop a lease from the in-flight set and free its capacity permit. This pair is the
- * single invariant "a permit is held iff its key is in-flight"; returns {@code true} if
- * this call performed the removal (so callers can log only a real eviction).
- */
- private boolean releaseLease(String key) {
- if (inFlight.remove(key) != null) {
+ private void finishAttempt(InFlight e) {
+ final String key = e.key();
+ activeSince.remove(key);
+ warned.remove(key);
+ if (active.remove(key) != null) {
slots.release();
- return true;
}
- return false;
+ if (closed && active.isEmpty() && heartbeat != null) {
+ heartbeat.shutdown();
+ }
}
/**
- * Heartbeat tick: renew every in-flight lease so an alive consumer keeps ownership,
- * and release the lease of any single invocation that has exceeded
- * {@link #maxProcessingTime()} (safety valve; does not interrupt the handler thread).
+ * Heartbeat every currently executing invocation. Long-running handlers are warned
+ * about, but never evicted while their thread remains active.
*/
private void heartbeatTick() {
final long now = System.currentTimeMillis();
final long maxMillis = maxProcessingTime().toMillis();
- for (InFlight e : inFlight.values()) {
+ for (InFlight e : active.values()) {
final String key = e.key();
- final long start = active.getOrDefault(key, now);
+ final long start = activeSince.getOrDefault(key, now);
if (now - start > maxMillis) {
- // a single invocation is stalled beyond the bound: stop renewing so the
- // lease becomes reclaimable, and free its capacity permit
- if (releaseLease(key)) {
- log.warn("Work queue - releasing lease of stalled entry={} after {} - reclaimable after visibility timeout",
+ if (warned.add(key)) {
+ log.warn("Work queue - handler still active for entry={} after {}; continuing lease renewal to prevent overlap",
key, Duration.ofMillis(now - start));
}
}
- else {
- try {
- queue.renewLease(e.queueId(), e.leaseId());
- }
- catch (Throwable t) {
- // swallow transient errors; the next tick retries
- log.warn("Work queue - error renewing lease for entry={} - cause: {}", key, t.getMessage());
- }
+ try {
+ queue.renewLease(e.queueId(), e.leaseId());
+ }
+ catch (Throwable t) {
+ // swallow transient errors; the next tick retries
+ log.warn("Work queue - error renewing lease for entry={} - cause: {}", key, t.getMessage());
}
}
}
/**
- * Shutdown orderly the queue: stop the dispatcher, cancel pending re-polls, drain
- * the worker pool so active handlers finish and ack, release any remaining leases so
- * they are redelivered, and finally stop the heartbeat daemon.
+ * Stop intake. Active handlers remain heartbeated until they finish; close never makes
+ * an entry reclaimable while its handler thread is still executing.
*/
@Override
public void close() {
@@ -612,26 +607,10 @@ public void close() {
catch (Exception e) {
log.debug("Unexpected error while terminating {} - cause: {}", name0, e.getMessage());
}
- // 2. cancel pending scheduled re-polls
- if (scheduler != null) {
- scheduler.shutdownNow();
- }
- // 3. the handler executor is shared / container-managed — not shut down here;
- // any active handler finishes on its own (short-lived) and acks
- // 4. release any lease still held so it is redelivered without waiting for lapse
- for (InFlight e : inFlight.values()) {
- if (inFlight.remove(e.key()) != null) {
- try {
- queue.release(e.queueId(), e.leaseId());
- }
- catch (Throwable t) {
- log.debug("Work queue - error releasing entry={} on shutdown - cause: {}", e.key(), t.getMessage());
- }
- }
- }
- // 5. stop the heartbeat daemon last (any remaining leases lapse -> peers reclaim)
- if (heartbeat != null) {
- heartbeat.shutdownNow();
+ // The handler executor is shared/container-managed. Keep the heartbeat daemon alive
+ // until every active invocation completes; finishAttempt shuts it down at that point.
+ if (heartbeat != null && active.isEmpty()) {
+ heartbeat.shutdown();
}
}
diff --git a/lib-data-workqueue/src/main/java/io/seqera/data/workqueue/LocalWorkQueue.java b/lib-data-workqueue/src/main/java/io/seqera/data/workqueue/LocalWorkQueue.java
index 5c1ea54c..0288b577 100644
--- a/lib-data-workqueue/src/main/java/io/seqera/data/workqueue/LocalWorkQueue.java
+++ b/lib-data-workqueue/src/main/java/io/seqera/data/workqueue/LocalWorkQueue.java
@@ -90,7 +90,7 @@ public void offer(String queueId, String message) {
* the lease id is simply the message value itself (used to re-offer it on release).
*/
@Override
- public Lease receive(String queueId) {
+ public Lease receiveNew(String queueId) {
final var message = delegate
.get(queueId)
.poll();
@@ -100,6 +100,11 @@ public Lease receive(String queueId) {
return new Lease<>(message, message);
}
+ @Override
+ public Lease reclaim(String queueId) {
+ return null;
+ }
+
/**
* {@inheritDoc}
*
diff --git a/lib-data-workqueue/src/main/java/io/seqera/data/workqueue/WorkQueue.java b/lib-data-workqueue/src/main/java/io/seqera/data/workqueue/WorkQueue.java
index d1d4e63f..5fd7215a 100644
--- a/lib-data-workqueue/src/main/java/io/seqera/data/workqueue/WorkQueue.java
+++ b/lib-data-workqueue/src/main/java/io/seqera/data/workqueue/WorkQueue.java
@@ -162,15 +162,26 @@ default boolean consume(String queueId, MessageConsumer consumer) {
record Lease(String id, M message) {}
/**
- * Receives one message (either newly delivered or reclaimed from a stalled consumer)
- * without acknowledging it. The caller becomes responsible for
- * eventually calling {@link #ack(String, String)} once processing terminates, or
- * {@link #release(String, String)} to hand it back for later redelivery.
- *
- * @param queueId the unique identifier of the source queue; must not be null or empty
- * @return a {@link Lease} for the delivered message, or {@code null} if none is available
+ * Receives one message that has never previously been delivered to the consumer group.
+ * The caller is responsible for acknowledging it or leaving it pending for retry.
+ */
+ Lease receiveNew(String queueId);
+
+ /**
+ * Reclaims one pending message whose owner has stopped renewing it for longer than the
+ * implementation's visibility timeout.
*/
- Lease receive(String queueId);
+ Lease reclaim(String queueId);
+
+ /**
+ * Receives one message, preferring an expired pending entry for compatibility with the
+ * original synchronous API. Asynchronous dispatchers should select explicitly between
+ * {@link #receiveNew(String)} and {@link #reclaim(String)} to provide intake fairness.
+ */
+ default Lease receive(String queueId) {
+ final Lease reclaimed = reclaim(queueId);
+ return reclaimed != null ? reclaimed : receiveNew(queueId);
+ }
/**
* Resets the idle time of the given lease (heartbeat), so that an alive consumer
@@ -216,12 +227,11 @@ default Duration heartbeatInterval() {
}
/**
- * Upper bound on a single {@code accept()} invocation before its lease is released
- * (safety valve); it does not interrupt the handler thread. Returns {@code null}
- * when the implementation has no lease concept, in which case the caller uses its
- * own default.
+ * Observability threshold after which an active {@code accept()} invocation is reported
+ * as stalled. It never expires or releases the lease while the handler is still running.
+ * Returns {@code null} when the implementation supplies no threshold.
*
- * @return the maximum single-invocation processing time, or {@code null}
+ * @return the warning threshold for one handler invocation, or {@code null}
*/
default Duration maxProcessingTime() {
return null;
diff --git a/lib-data-workqueue/src/test/groovy/io/seqera/data/workqueue/AsyncWorkQueueLocalTest.groovy b/lib-data-workqueue/src/test/groovy/io/seqera/data/workqueue/AsyncWorkQueueLocalTest.groovy
index 8884c125..04412d47 100644
--- a/lib-data-workqueue/src/test/groovy/io/seqera/data/workqueue/AsyncWorkQueueLocalTest.groovy
+++ b/lib-data-workqueue/src/test/groovy/io/seqera/data/workqueue/AsyncWorkQueueLocalTest.groovy
@@ -31,7 +31,7 @@ import spock.util.concurrent.PollingConditions
* Async-processing behaviour of {@link AbstractWorkQueue} exercised over the
* in-memory {@link LocalWorkQueue} backend, so these run WITHOUT Docker.
*
- * Covers: non-blocking dispatch, concurrency, re-poll cadence, serial-per-command,
+ * Covers: non-blocking dispatch, invocation concurrency, local retry, serial-per-message,
* backpressure, and concurrency==1 default.
*
* @author Paolo Di Tommaso
@@ -87,8 +87,8 @@ class AsyncWorkQueueLocalTest extends Specification {
queue.close()
}
- // a not-yet-terminal command is re-invoked at ~pollInterval (Model B)
- def 'should re-poll a not-yet-terminal command at poll interval' () {
+ // Local queues have no PEL/visibility timeout, so RETRY is immediately re-offered.
+ def 'should release an invocation slot before retrying a non-terminal message' () {
given:
def poll = Duration.ofMillis(300)
def target = new LocalWorkQueue()
@@ -103,16 +103,12 @@ class AsyncWorkQueueLocalTest extends Specification {
return timestamps.size() >= 5
})
queue.offer(id, 'running')
+ queue.offer(id, 'other')
then:
new PollingConditions(timeout: 10).eventually {
- assert timestamps.size() == 5
+ assert timestamps.size() >= 5
}
- and:
- def times = timestamps.toList()
- def gaps = (1..= 150 && it <= 1_500 }
cleanup:
queue.close()
@@ -219,7 +215,8 @@ class AsyncWorkQueueLocalTest extends Specification {
def target = [
init : { String q -> },
offer : { String q, String m -> },
- receive : { String q -> deliveries.getAndIncrement() < 2 ? new WorkQueue.Lease('dup-id', 'payload') : null },
+ receiveNew: { String q -> deliveries.getAndIncrement() < 2 ? new WorkQueue.Lease('dup-id', 'payload') : null },
+ reclaim : { String q -> null },
renewLease: { String q, String id -> },
ack : { String q, String id -> acks.incrementAndGet() },
release : { String q, String id -> },
diff --git a/lib-data-workqueue/src/test/groovy/io/seqera/data/workqueue/TunableQueue.groovy b/lib-data-workqueue/src/test/groovy/io/seqera/data/workqueue/TunableQueue.groovy
index 4e2523a9..431fbaf7 100644
--- a/lib-data-workqueue/src/test/groovy/io/seqera/data/workqueue/TunableQueue.groovy
+++ b/lib-data-workqueue/src/test/groovy/io/seqera/data/workqueue/TunableQueue.groovy
@@ -24,7 +24,7 @@ import io.seqera.serde.encode.StringEncodingStrategy
/**
* A {@link AbstractWorkQueue} used by the async-processing tests. It carries a
* String payload (identity encoding) and exposes the async knobs — concurrency,
- * poll interval, heartbeat interval and max-processing-time — as constructor options
+ * poll interval, heartbeat interval and handler-warning threshold — as constructor options
* so each test can tune them independently.
*
* @author Paolo Di Tommaso