Move job queues to lib-data-workqueue, fix duplicate job processing and drain on shutdown

build.gradle

@@ -73,7 +73,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/install/reference.md

@@ -425,6 +425,9 @@ Configure Kubernetes-specific settings for Wave. Build and scan processes share
 
   </div>
 
+`wave.build.k8s.read-timeout` *(optional)*
+: Maximum time to wait for data from the Kubernetes API server on an open connection before the request fails (default: `50s`).
+
 `wave.build.k8s.resources.requests.cpu` *(optional)*
 : [CPU resources](https://kubernetes.io/docs/concepts/configuration/manage-resources-containers/#resource-units-in-kubernetes) allocated to Wave build processes.
   For example, set to `2` (2 CPU cores) or `1500Mi` (1.5 CPU cores).
@@ -745,6 +748,10 @@ Configure the connection to the Seqera license management server:
 
 Configure polling and scheduling intervals for Wave's asynchronous job processing.
 
+`wave.job-manager.drain-timeout` *(optional)*
+: Maximum time the application shutdown waits for the jobs being launched or checked for completion. Jobs not completed within this time are handed over to another replica (default: `15s`).
+  It must be shorter than `micronaut.lifecycle.graceful-shutdown.grace-period` (default: `20s`).
+
 `wave.job-manager.max-running-jobs` *(optional)*
 : Maximum number of build, scan, and mirror jobs Wave runs concurrently (default: `20`).
   Pair this with a `ResourceQuota` on the build namespace to bound build resource usage.
@@ -755,15 +762,20 @@ 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 backed work queue used for internal job processing.
+
+Each setting falls back to the legacy `wave.message-stream.*` property it replaces, so a deployment configured for the previous message-stream implementation keeps its values after upgrading with no configuration change.
 
-Configure how Wave consumes messages from the Redis stream used for internal event processing.
+`wave.work-queue.consumer-group-name` *(optional)*
+: Name of the Redis consumer group used to read the work queues. Falls back to `wave.message-stream.consumer-group-name` (default: `wave-message-stream`).
 
-`wave.message-stream.claim-timeout` *(optional)*
-: Timeout for claiming messages from the Redis stream (default: `5s`).
+`wave.work-queue.visibility-timeout` *(optional)*
+: How long a claimed message stays invisible to other consumers before it is redelivered. Falls back to `wave.message-stream.claim-timeout` (default: `45s`).
 
-`wave.message-stream.consume-warn-timeout` *(optional)*
-: Threshold duration after which a slow message consumer triggers a warning (default: `4s`).
+`wave.work-queue.consumer-warn-timeout` *(optional)*
+: Threshold duration after which a slow message consumer triggers a warning. Falls back to `wave.message-stream.consume-warn-timeout` (default: `45s`).
 
 ## Thread monitor
 

src/main/groovy/io/seqera/wave/configuration/JobManagerConfig.groovy

@@ -49,4 +49,12 @@ class JobManagerConfig {
 
     @Value('${wave.job-manager.max-running-jobs:20}')
     int maxRunningJobs
+
+    /**
+     * How long the application shutdown waits for the jobs being processed. Keep it below
+     * {@code micronaut.lifecycle.graceful-shutdown.grace-period}: when that expires Micronaut
+     * proceeds to destroy the beans regardless
+     */
+    @Value('${wave.job-manager.drain-timeout:15s}')
+    Duration drainTimeout
 }

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,89 @@
+/*
+ *  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 (each falls back to the legacy {@code wave.message-stream.*}
+ * property it replaces, then to the default shown):
+ * <ul>
+ *   <li>{@code wave.work-queue.consumer-group-name} - Name for the Redis consumer group
+ *       (falls back to {@code wave.message-stream.consumer-group-name}; 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 (falls back to {@code wave.message-stream.claim-timeout}; default: 45s)</li>
+ *   <li>{@code wave.work-queue.consumer-warn-timeout} - Timeout threshold for consumer warnings
+ *       (falls back to {@code wave.message-stream.consume-warn-timeout}; default: 45s)</li>
+ * </ul>
+ *
+ * <p>The fallback to the legacy {@code message-stream} keys means a deployment configured for the
+ * message-stream implementation keeps its values after the work-queue upgrade with no config change;
+ * the final defaults (45s) match the current production deployment.
+ *
+ * <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 {
+
+    // Each setting falls back to the legacy `wave.message-stream.*` property it replaces, so
+    // a deployment configured for the message-stream implementation keeps its values after the
+    // work-queue upgrade without any config change. The final defaults match the current
+    // production deployment (see platform-deployment).
+    @Value('${wave.work-queue.consumer-group-name:wave.message-stream.consumer-group-name:wave-message-stream}')
+    String defaultConsumerGroupName
+
+    @Value('${wave.work-queue.visibility-timeout:wave.message-stream.claim-timeout:45s}')
+    Duration visibilityTimeout
+
+    @Value('${wave.work-queue.consumer-warn-timeout:wave.message-stream.consume-warn-timeout:45s}')
+    Duration consumerWarnTimeout
+
+}

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

@@ -20,6 +20,8 @@ package io.seqera.wave.service.job
 
 import java.time.Duration
 import java.time.Instant
+import java.util.concurrent.CompletableFuture
+import java.util.concurrent.CompletionStage
 import java.util.concurrent.ExecutorService
 
 import com.github.benmanes.caffeine.cache.Cache
@@ -28,7 +30,10 @@ import groovy.transform.CompileStatic
 import groovy.util.logging.Slf4j
 import io.micronaut.context.annotation.Context
 import io.micronaut.context.annotation.Requires
+import io.micronaut.runtime.graceful.GracefulShutdownCapable
 import io.micronaut.scheduling.TaskExecutors
+import io.seqera.data.workqueue.MessageConsumer.Decision
+import io.seqera.data.workqueue.MessageLease
 import io.seqera.util.trace.TraceElapsedTime
 import io.seqera.wave.configuration.JobManagerConfig
 import io.seqera.wave.configuration.WaveLite
@@ -44,7 +49,7 @@ import jakarta.inject.Named
 @Context
 @Requires(missingBeans = WaveLite)
 @CompileStatic
-class JobManager {
+class JobManager implements GracefulShutdownCapable {
 
     @Inject
     private JobService jobService
@@ -75,12 +80,51 @@ 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)-> {
+            keepAlive(lease)
+            return launchJob(job) ? Decision.ACK : Decision.RETRY
+        })
+        processingQueue.addConsumer((job, lease)-> {
+            keepAlive(lease)
+            return processJob(job) ? Decision.ACK : Decision.RETRY
+        })
+    }
+
+    /**
+     * Keep the lease renewed for as long as the consumer runs, so a slow job launch or status
+     * check is never redelivered to another replica while it is still in progress. The probe
+     * can be constant because the consumer is synchronous: the lease is registered before the
+     * consumer is invoked and released when it returns
+     */
+    protected static void keepAlive(MessageLease lease) {
+        lease.bindLiveness(() -> true)
+    }
+
+    /**
+     * Drain the job queues when the application shutdown begins. Micronaut invokes this
+     * before destroying any bean, so the jobs being launched or checked can complete while
+     * the services they depend on are still available. Requires
+     * {@code micronaut.lifecycle.graceful-shutdown.enabled=true}.
+     *
+     * <p>Jobs not completed within the drain timeout keep their queue entry and are
+     * redelivered to another replica once their lease expires.
+     */
+    @Override
+    CompletionStage<?> shutdownGracefully() {
+        log.info "Draining job queues - timeout=${config.drainTimeout}"
+        // close both queues in parallel, Micronaut invokes each shutdown hook on the same thread
+        return CompletableFuture.allOf(
+                CompletableFuture.runAsync(() -> pendingQueue.close(config.drainTimeout)),
+                CompletableFuture.runAsync(() -> processingQueue.close(config.drainTimeout)) )
     }
 
     @TraceElapsedTime(thresholdMillis = '${wave.trace.k8s.threshold:500}')
     protected boolean launchJob0(JobSpec job) {
+        // the processing queue is shared by all replicas and holds one entry per running job, so
+        // `maxRunningJobs` is a cluster-wide limit, not a per-replica one. It is a soft limit: the
+        // length is read before launching, so replicas launching concurrently can overshoot it
         final processingQueueLen = processingQueue.length()
         final canLaunchNewJobs = processingQueueLen < config.maxRunningJobs
         if( canLaunchNewJobs ) {