target) {
super(target)
}
/**
* Creates an instance of the required {@link StringEncodingStrategy} to serialize
* and deserialize message events to/from JSON format.
- *
+ *
* 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.
- *
+ * representations for queueing operations.
+ *
* @return a new instance of {@link StringEncodingStrategy} configured for type {@code M}
*/
@Override
diff --git a/src/main/groovy/io/seqera/wave/service/data/workqueue/RedisWorkQueueConfigBean.groovy b/src/main/groovy/io/seqera/wave/service/data/workqueue/RedisWorkQueueConfigBean.groovy
new file mode 100644
index 0000000000..19c8193e7a
--- /dev/null
+++ b/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 .
+ */
+
+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.
+ *
+ * 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.
+ *
+ *
Configuration properties:
+ *
+ * - {@code wave.work-queue.consumer-group-name} - Name for the Redis consumer group (default: "wave-message-stream")
+ * - {@code wave.work-queue.visibility-timeout} - How long a claimed message stays invisible to other consumers before being redelivered (default: 5s)
+ * - {@code wave.work-queue.consumer-warn-timeout} - Timeout threshold for consumer warnings (default: 4s)
+ *
+ *
+ * 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.
+ *
+ *
Example application configuration:
+ *
+ * # application.yml
+ * wave:
+ * work-queue:
+ * consumer-group-name: "wave-prod-queue"
+ * visibility-timeout: "10s"
+ * consumer-warn-timeout: "8s"
+ *
+ *
+ * This bean implements {@link RedisWorkQueueConfig} to provide configuration
+ * values to Redis work queue components throughout the application.
+ *
+ * @author Paolo Di Tommaso
+ * @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
+
+}
diff --git a/src/main/groovy/io/seqera/wave/service/job/JobManager.groovy b/src/main/groovy/io/seqera/wave/service/job/JobManager.groovy
index 9865f6b7d9..69aa36b822 100644
--- a/src/main/groovy/io/seqera/wave/service/job/JobManager.groovy
+++ b/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}')
diff --git a/src/main/groovy/io/seqera/wave/service/job/JobPendingQueue.groovy b/src/main/groovy/io/seqera/wave/service/job/JobPendingQueue.groovy
index 5b0712de98..ed99dd5232 100644
--- a/src/main/groovy/io/seqera/wave/service/job/JobPendingQueue.groovy
+++ b/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 {
+class JobPendingQueue extends BaseWorkQueue {
- private final static String STREAM_NAME = 'jobs-pending/v2'
-
- private StringEncodingStrategy encoder
+ private final static String QUEUE_ID = 'jobs-pending/v2'
private JobManagerConfig config
- JobPendingQueue(MessageStream target, JobManagerConfig config) {
+ JobPendingQueue(WorkQueue target, JobManagerConfig config) {
super(target)
- this.encoder = new MoshiEncodeStrategy() {}
this.config = config
log.info "Created jobs pending queue - config=${config}"
}
@@ -68,15 +63,15 @@ class JobPendingQueue extends BaseMessageStream {
}
final void submit(JobSpec jobSpec) {
- super.offer(STREAM_NAME, jobSpec)
+ super.offer(QUEUE_ID, jobSpec)
}
final void addConsumer(MessageConsumer 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
diff --git a/src/main/groovy/io/seqera/wave/service/job/JobProcessingQueue.groovy b/src/main/groovy/io/seqera/wave/service/job/JobProcessingQueue.groovy
index 502bb9eeef..d6aa0ac703 100644
--- a/src/main/groovy/io/seqera/wave/service/job/JobProcessingQueue.groovy
+++ b/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 {
+class JobProcessingQueue extends BaseWorkQueue {
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 target, JobManagerConfig config) {
+ JobProcessingQueue(WorkQueue target, JobManagerConfig config) {
super(target)
this.config = config
log.info "Created jobs processing queue - config=${config}"
@@ -64,15 +64,15 @@ class JobProcessingQueue extends BaseMessageStream {
}
final void offer(JobSpec jobSpec) {
- super.offer(STREAM_NAME, jobSpec)
+ super.offer(QUEUE_ID, jobSpec)
}
final void addConsumer(MessageConsumer 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