Replace lib-data-stream-redis with lib-data-workqueue

build.gradle

@@ -66,7 +66,8 @@ dependencies {
     implementation 'io.seqera:lib-crypto:1.0.0'
     implementation 'io.seqera:jedis-lock:1.0.0'
     implementation 'io.seqera:lib-data-queue-redis:1.1.2'
-    implementation 'io.seqera:lib-data-stream-redis:1.2.0'
+    implementation 'io.seqera:lib-data-workqueue:2.0.0'
+    implementation 'io.seqera:lib-data-workqueue-redis:2.0.0'
     implementation 'io.seqera:lib-data-range-redis:1.1.0'
     implementation 'io.seqera:lib-jedis-pool:1.0.0'
     implementation 'io.micronaut:micronaut-http-client'

docs/configuration.md

@@ -587,14 +587,17 @@ Configure polling and scheduling intervals for Wave's asynchronous job processin
 `wave.job-manager.scheduler-interval` *(optional)*
 : Interval for the job scheduler to process queued jobs (default: `1s`).
 
-## Message stream
+## Work queue
 
-Configure how Wave consumes messages from the Redis stream used for internal event processing.
+Configure how Wave consumes messages from the Redis backed work queue used for internal job processing.
 
-`wave.message-stream.claim-timeout` *(optional)*
-: Timeout for claiming messages from the Redis stream (default: `5s`).
+`wave.work-queue.consumer-group-name` *(optional)*
+: Name of the Redis consumer group used to read the work queues (default: `wave-message-stream`).
 
-`wave.message-stream.consume-warn-timeout` *(optional)*
+`wave.work-queue.visibility-timeout` *(optional)*
+: How long a claimed message stays invisible to other consumers before it is redelivered (default: `5s`).
+
+`wave.work-queue.consumer-warn-timeout` *(optional)*
 : Threshold duration after which a slow message consumer triggers a warning (default: `4s`).
 
 ## Thread monitor

src/main/groovy/io/seqera/wave/service/data/stream/BaseMessageStream.groovy → src/main/groovy/io/seqera/wave/service/data/workqueue/BaseWorkQueue.groovy

@@ -16,50 +16,52 @@
  *  along with this program.  If not, see <https://www.gnu.org/licenses/>.
  */
 
-package io.seqera.wave.service.data.stream
+package io.seqera.wave.service.data.workqueue
 
-import io.seqera.data.stream.AbstractMessageStream
-import io.seqera.data.stream.MessageStream
+import io.seqera.data.workqueue.AbstractWorkQueue
+import io.seqera.data.workqueue.WorkQueue
 import io.seqera.lang.type.TypeHelper
 import io.seqera.serde.encode.StringEncodingStrategy
 import io.seqera.serde.moshi.MoshiEncodeStrategy
 
 /**
- * Base abstract class for implementing message streams in the Wave application.
- * 
- * <p>This class extends {@link AbstractMessageStream} and provides a foundation for
- * creating type-safe message streams with automatic JSON serialization/deserialization
+ * Base abstract class for implementing work queues in the Wave application.
+ *
+ * <p>This class extends {@link AbstractWorkQueue} and provides a foundation for
+ * creating type-safe work queues with automatic JSON serialization/deserialization
  * using the Moshi library. It handles the encoding strategy configuration and
- * provides a consistent interface for message stream implementations.</p>
- * 
- * <p>Message streams are used for real-time message processing and event handling,
- * allowing for continuous data flow and stream-based operations. Concrete implementations
- * should extend this class to define specific streaming behavior and message processing logic.</p>
- * 
- * @param <M> the type of messages that this stream will handle
+ * provides a consistent interface for work queue implementations.</p>
+ *
+ * <p>Work queues deliver each message to exactly one consumer at a time under a
+ * message lease: a message is removed from the queue only when the consumer settles
+ * it as acknowledged, and is otherwise redelivered once the lease expires. Concrete
+ * implementations should extend this class to define specific queueing behaviour and
+ * message processing logic.</p>
+ *
+ * @param <M> the type of messages that this queue will handle
  * @author Paolo Di Tommaso <paolo.ditommaso@gmail.com>
  */
-abstract class BaseMessageStream<M> extends AbstractMessageStream<M> {
+abstract class BaseWorkQueue<M> extends AbstractWorkQueue<M> {
 
     /**
-     * Constructs a new BaseMessageStream with the specified target stream.
-     * 
-     * @param target the underlying string-based message stream that handles
-     *               the actual message transport and streaming operations
+     * Constructs a new BaseWorkQueue with the specified target queue.
+     *
+     * @param target the underlying string-based work queue that handles
+     *               the actual message transport and lease management
      */
-    BaseMessageStream(MessageStream<String> target) {
+    BaseWorkQueue(WorkQueue<String> target) {
         super(target)
     }
 
     /**
      * Creates an instance of the required {@link StringEncodingStrategy} to serialize
      * and deserialize message events to/from JSON format.
-     * 
+     *
      * <p>This method uses reflection to determine the generic type parameter {@code M}
      * and creates a Moshi-based encoding strategy that can handle the automatic
      * conversion between the strongly-typed message objects and their JSON string
-     * representations for streaming operations.</p>
-     * 
+     * representations for queueing operations.</p>
+     *
      * @return a new instance of {@link StringEncodingStrategy} configured for type {@code M}
      */
     @Override

src/main/groovy/io/seqera/wave/service/data/workqueue/RedisWorkQueueConfigBean.groovy

@@ -0,0 +1,77 @@
+/*
+ *  Wave, containers provisioning service
+ *  Copyright (c) 2023-2025, Seqera Labs
+ *
+ *  This program is free software: you can redistribute it and/or modify
+ *  it under the terms of the GNU Affero General Public License as published by
+ *  the Free Software Foundation, either version 3 of the License, or
+ *  (at your option) any later version.
+ *
+ *  This program is distributed in the hope that it will be useful,
+ *  but WITHOUT ANY WARRANTY; without even the implied warranty of
+ *  MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+ *  GNU Affero General Public License for more details.
+ *
+ *  You should have received a copy of the GNU Affero General Public License
+ *  along with this program.  If not, see <https://www.gnu.org/licenses/>.
+ */
+
+package io.seqera.wave.service.data.workqueue
+
+import java.time.Duration
+
+import groovy.transform.CompileStatic
+import io.micronaut.context.annotation.Value
+import io.seqera.data.workqueue.redis.RedisWorkQueueConfig
+import jakarta.inject.Singleton
+
+/**
+ * Configuration bean for the Redis backed work queue in Wave application.
+ *
+ * <p>This configuration provides centralized settings for Redis work queue
+ * components, including consumer group management and lease timings. The
+ * configuration values are injected from application properties with sensible
+ * defaults.
+ *
+ * <p>Configuration properties:
+ * <ul>
+ *   <li>{@code wave.work-queue.consumer-group-name} - Name for the Redis consumer group (default: "wave-message-stream")</li>
+ *   <li>{@code wave.work-queue.visibility-timeout} - How long a claimed message stays invisible to other consumers before being redelivered (default: 5s)</li>
+ *   <li>{@code wave.work-queue.consumer-warn-timeout} - Timeout threshold for consumer warnings (default: 4s)</li>
+ * </ul>
+ *
+ * <p>Note the consumer group name default is intentionally {@code wave-message-stream}:
+ * the work queue reuses the Redis streams and consumer group created by the message
+ * stream implementation it replaces, so messages queued before an upgrade are still
+ * delivered afterwards.
+ *
+ * <p>Example application configuration:
+ * <pre>
+ * # application.yml
+ * wave:
+ *   work-queue:
+ *     consumer-group-name: "wave-prod-queue"
+ *     visibility-timeout: "10s"
+ *     consumer-warn-timeout: "8s"
+ * </pre>
+ *
+ * <p>This bean implements {@link RedisWorkQueueConfig} to provide configuration
+ * values to Redis work queue components throughout the application.
+ *
+ * @author Paolo Di Tommaso <paolo.ditommaso@gmail.com>
+ * @see io.seqera.data.workqueue.redis.RedisWorkQueueConfig
+ */
+@CompileStatic
+@Singleton
+class RedisWorkQueueConfigBean implements RedisWorkQueueConfig {
+
+    @Value('${wave.work-queue.consumer-group-name:wave-message-stream}')
+    String defaultConsumerGroupName
+
+    @Value('${wave.work-queue.visibility-timeout:5s}')
+    Duration visibilityTimeout
+
+    @Value('${wave.work-queue.consumer-warn-timeout:4s}')
+    Duration consumerWarnTimeout
+
+}

src/main/groovy/io/seqera/wave/service/job/JobManager.groovy

@@ -29,6 +29,7 @@ import groovy.util.logging.Slf4j
 import io.micronaut.context.annotation.Context
 import io.micronaut.context.annotation.Requires
 import io.micronaut.scheduling.TaskExecutors
+import io.seqera.data.workqueue.MessageConsumer.Decision
 import io.seqera.util.trace.TraceElapsedTime
 import io.seqera.wave.configuration.JobManagerConfig
 import io.seqera.wave.configuration.WaveLite
@@ -75,8 +76,10 @@ class JobManager {
                 .expireAfterWrite(config.graceInterval.multipliedBy(2))
                 .executor(ioExecutor)
                 .build()
-        pendingQueue.addConsumer((job)-> launchJob(job))
-        processingQueue.addConsumer((job)-> processJob(job))
+        // a `false` outcome settles the message as RETRY, leaving it queued for
+        // redelivery once the lease expires; `true` acknowledges and removes it
+        pendingQueue.addConsumer((job, lease)-> launchJob(job) ? Decision.ACK : Decision.RETRY)
+        processingQueue.addConsumer((job, lease)-> processJob(job) ? Decision.ACK : Decision.RETRY)
     }
 
     @TraceElapsedTime(thresholdMillis = '${wave.trace.k8s.threshold:500}')

src/main/groovy/io/seqera/wave/service/job/JobPendingQueue.groovy

@@ -23,13 +23,11 @@ import java.time.Duration
 import groovy.transform.CompileStatic
 import groovy.util.logging.Slf4j
 import io.micronaut.context.annotation.Requires
-import io.seqera.data.stream.MessageConsumer
-import io.seqera.data.stream.MessageStream
-import io.seqera.serde.encode.StringEncodingStrategy
+import io.seqera.data.workqueue.MessageConsumer
+import io.seqera.data.workqueue.WorkQueue
 import io.seqera.wave.configuration.JobManagerConfig
 import io.seqera.wave.configuration.WaveLite
-import io.seqera.serde.moshi.MoshiEncodeStrategy
-import io.seqera.wave.service.data.stream.BaseMessageStream
+import io.seqera.wave.service.data.workqueue.BaseWorkQueue
 import jakarta.annotation.PreDestroy
 import jakarta.inject.Singleton
 /**
@@ -42,17 +40,14 @@ import jakarta.inject.Singleton
 @Slf4j
 @Singleton
 @CompileStatic
-class JobPendingQueue extends BaseMessageStream<JobSpec> {
+class JobPendingQueue extends BaseWorkQueue<JobSpec> {
 
-    private final static String STREAM_NAME = 'jobs-pending/v2'
-
-    private StringEncodingStrategy<JobSpec> encoder
+    private final static String QUEUE_ID = 'jobs-pending/v2'
 
     private JobManagerConfig config
 
-    JobPendingQueue(MessageStream<String> target, JobManagerConfig config) {
+    JobPendingQueue(WorkQueue<String> target, JobManagerConfig config) {
         super(target)
-        this.encoder = new MoshiEncodeStrategy<JobSpec>() {}
         this.config = config
         log.info "Created jobs pending queue - config=${config}"
     }
@@ -68,15 +63,15 @@ class JobPendingQueue extends BaseMessageStream<JobSpec> {
     }
 
     final void submit(JobSpec jobSpec) {
-        super.offer(STREAM_NAME, jobSpec)
+        super.offer(QUEUE_ID, jobSpec)
     }
 
     final void addConsumer(MessageConsumer<JobSpec> consumer) {
-        super.addConsumer(STREAM_NAME, consumer)
+        super.addConsumer(QUEUE_ID, consumer)
     }
 
     final int length() {
-        return super.length(STREAM_NAME)
+        return super.length(QUEUE_ID)
     }
 
     @PreDestroy

src/main/groovy/io/seqera/wave/service/job/JobProcessingQueue.groovy

@@ -23,11 +23,11 @@ import java.time.Duration
 import groovy.transform.CompileStatic
 import groovy.util.logging.Slf4j
 import io.micronaut.context.annotation.Requires
-import io.seqera.data.stream.MessageConsumer
-import io.seqera.data.stream.MessageStream
+import io.seqera.data.workqueue.MessageConsumer
+import io.seqera.data.workqueue.WorkQueue
 import io.seqera.wave.configuration.JobManagerConfig
 import io.seqera.wave.configuration.WaveLite
-import io.seqera.wave.service.data.stream.BaseMessageStream
+import io.seqera.wave.service.data.workqueue.BaseWorkQueue
 import jakarta.annotation.PreDestroy
 import jakarta.inject.Singleton
 /**
@@ -39,15 +39,15 @@ import jakarta.inject.Singleton
 @Slf4j
 @Singleton
 @CompileStatic
-class JobProcessingQueue extends BaseMessageStream<JobSpec> {
+class JobProcessingQueue extends BaseWorkQueue<JobSpec> {
 
     private final static String QUEUE_NAME = "jobs-queue"
 
-    private final static String STREAM_NAME = "jobs-queue/v1"
+    private final static String QUEUE_ID = "jobs-queue/v1"
 
     private final JobManagerConfig config
 
-    JobProcessingQueue(MessageStream<String> target, JobManagerConfig config) {
+    JobProcessingQueue(WorkQueue<String> target, JobManagerConfig config) {
         super(target)
         this.config = config
         log.info "Created jobs processing queue - config=${config}"
@@ -64,15 +64,15 @@ class JobProcessingQueue extends BaseMessageStream<JobSpec> {
     }
 
     final void offer(JobSpec jobSpec) {
-        super.offer(STREAM_NAME, jobSpec)
+        super.offer(QUEUE_ID, jobSpec)
     }
 
     final void addConsumer(MessageConsumer<JobSpec> consumer) {
-        super.addConsumer(STREAM_NAME, consumer)
+        super.addConsumer(QUEUE_ID, consumer)
     }
 
     final int length() {
-        return super.length(STREAM_NAME)
+        return super.length(QUEUE_ID)
     }
 
     @PreDestroy