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..d1eccb08e0
--- /dev/null
+++ b/src/main/groovy/io/seqera/wave/service/data/workqueue/RedisWorkQueueConfigBean.groovy
@@ -0,0 +1,90 @@
+/*
+ * 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 (each falls back to the legacy {@code wave.message-stream.*}
+ * property it replaces, then to the default shown):
+ *
+ * - {@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")
+ * - {@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: 20s)
+ * - {@code wave.work-queue.consumer-warn-timeout} - Timeout threshold for consumer warnings
+ * (falls back to {@code wave.message-stream.consume-warn-timeout}; default: 45s)
+ *
+ *
+ * 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 visibility timeout default (20s) bounds the redelivery delay of a job re-check, while leaving
+ * room for missed lease renewals (renewed every visibility timeout / 4).
+ *
+ *
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 {
+
+ // 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:20s}')
+ Duration visibilityTimeout
+
+ @Value('${wave.work-queue.consumer-warn-timeout:wave.message-stream.consume-warn-timeout:45s}')
+ 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..90162999cd 100644
--- a/src/main/groovy/io/seqera/wave/service/job/JobManager.groovy
+++ b/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}.
+ *
+ * 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 ) {
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
diff --git a/src/main/groovy/io/seqera/wave/service/k8s/K8sClusterClient.groovy b/src/main/groovy/io/seqera/wave/service/k8s/K8sClusterClient.groovy
index 116bb37d5c..3c28c13ef7 100644
--- a/src/main/groovy/io/seqera/wave/service/k8s/K8sClusterClient.groovy
+++ b/src/main/groovy/io/seqera/wave/service/k8s/K8sClusterClient.groovy
@@ -18,11 +18,14 @@
package io.seqera.wave.service.k8s
+import java.time.Duration
+
import groovy.transform.CompileStatic
import groovy.transform.Memoized
import io.kubernetes.client.openapi.ApiClient
import io.kubernetes.client.util.ClientBuilder
import io.micronaut.context.annotation.Requires
+import io.micronaut.context.annotation.Value
import jakarta.inject.Singleton
/**
@@ -38,6 +41,9 @@ import jakarta.inject.Singleton
@Singleton
class K8sClusterClient implements K8sClient {
+ @Value('${wave.build.k8s.read-timeout:50s}')
+ private Duration readTimeout
+
@Override
@Memoized
ApiClient apiClient() {
@@ -46,6 +52,11 @@ class K8sClusterClient implements K8sClient {
// 2. service-account bearer-token
// 3. service-account namespace
// 4. master endpoints(ip, port) from pre-set environment variables
- return ClientBuilder.cluster().build();
+ // the read timeout is set explicitly because the client default is none (`0`),
+ // which lets a stalled API server connection block the caller indefinitely
+ return ClientBuilder
+ .cluster()
+ .setReadTimeout(readTimeout)
+ .build()
}
}
diff --git a/src/main/groovy/io/seqera/wave/service/k8s/K8sConfigClient.groovy b/src/main/groovy/io/seqera/wave/service/k8s/K8sConfigClient.groovy
index b127d01efc..b84e1e0f5a 100644
--- a/src/main/groovy/io/seqera/wave/service/k8s/K8sConfigClient.groovy
+++ b/src/main/groovy/io/seqera/wave/service/k8s/K8sConfigClient.groovy
@@ -18,6 +18,8 @@
package io.seqera.wave.service.k8s
+import java.time.Duration
+
import io.micronaut.core.annotation.Nullable
import groovy.transform.CompileStatic
@@ -52,9 +54,12 @@ class K8sConfigClient implements K8sClient {
@Value('${wave.build.k8s.config-path}')
private String kubeConfigPath
+ @Value('${wave.build.k8s.read-timeout:50s}')
+ private Duration readTimeout
+
@Memoized
ApiClient apiClient() {
- log.info "Creating K8s config with path: $kubeConfigPath -- context: '$context'"
+ log.info "Creating K8s config with path: $kubeConfigPath -- context: '$context'; read-timeout: $readTimeout"
// load config
final config = KubeConfig.loadKubeConfig(new FileReader(kubeConfigPath))
@@ -65,6 +70,7 @@ class K8sConfigClient implements K8sClient {
// loading the out-of-cluster config, a kubeconfig from file-system
return ClientBuilder
.kubeconfig(config)
+ .setReadTimeout(readTimeout)
.build()
}
diff --git a/src/main/resources/application.yml b/src/main/resources/application.yml
index ebb2da0e15..f6afb8db44 100644
--- a/src/main/resources/application.yml
+++ b/src/main/resources/application.yml
@@ -14,6 +14,13 @@ micronaut:
# it is unused at the moment by the wave project. See https://micronaut-projects.github.io/micronaut-discovery-client/latest/guide/#introduction
discovery-client :
enabled : false
+ lifecycle:
+ # drain the job queues and in-flight HTTP requests before the application beans are destroyed.
+ # the grace period must exceed `wave.job-manager.drain-timeout` (15s) and stay within the pod
+ # `terminationGracePeriodSeconds` (30s by default), leaving time for the beans teardown
+ graceful-shutdown:
+ enabled: true
+ grace-period: 20s
server:
port: "${WAVE_SERVER_PORT:9090}"
# https://docs.micronaut.io/latest/guide/#atBlocking
diff --git a/src/test/groovy/io/seqera/wave/service/job/JobManagerTest.groovy b/src/test/groovy/io/seqera/wave/service/job/JobManagerTest.groovy
index a4345e896a..34b31a9b7d 100644
--- a/src/test/groovy/io/seqera/wave/service/job/JobManagerTest.groovy
+++ b/src/test/groovy/io/seqera/wave/service/job/JobManagerTest.groovy
@@ -18,12 +18,15 @@
package io.seqera.wave.service.job
+import io.seqera.data.workqueue.MessageLease
import io.seqera.wave.configuration.JobManagerConfig
import spock.lang.Specification
import java.nio.file.Path
import java.time.Duration
import java.time.Instant
+import java.util.concurrent.TimeUnit
+import java.util.function.BooleanSupplier
import com.github.benmanes.caffeine.cache.Cache
import com.github.benmanes.caffeine.cache.Caffeine
@@ -175,4 +178,32 @@ class JobManagerTest extends Specification {
cache.getIfPresent('1') == null
}
+
+ def 'should keep the lease alive while the consumer runs'() {
+ given:
+ def lease = Mock(MessageLease)
+ BooleanSupplier probe = null
+
+ when:
+ JobManager.keepAlive(lease)
+ then:
+ 1 * lease.bindLiveness(_) >> { args -> probe = args[0] as BooleanSupplier }
+ and:
+ probe.getAsBoolean()
+ }
+
+ def 'should close the job queues on graceful shutdown'() {
+ given:
+ def pendingQueue = Mock(JobPendingQueue)
+ def processingQueue = Mock(JobProcessingQueue)
+ def config = new JobManagerConfig(drainTimeout: Duration.ofSeconds(5))
+ def manager = new JobManager(config: config, pendingQueue: pendingQueue, processingQueue: processingQueue)
+
+ when:
+ manager.shutdownGracefully().toCompletableFuture().get(5, TimeUnit.SECONDS)
+
+ then:
+ 1 * pendingQueue.close(Duration.ofSeconds(5))
+ 1 * processingQueue.close(Duration.ofSeconds(5))
+ }
}
diff --git a/src/test/groovy/io/seqera/wave/service/k8s/K8sConfigClientTest.groovy b/src/test/groovy/io/seqera/wave/service/k8s/K8sConfigClientTest.groovy
new file mode 100644
index 0000000000..a15d1bcdcc
--- /dev/null
+++ b/src/test/groovy/io/seqera/wave/service/k8s/K8sConfigClientTest.groovy
@@ -0,0 +1,65 @@
+/*
+ * Wave, containers provisioning service
+ * Copyright (c) 2023-2024, 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.k8s
+
+import spock.lang.Specification
+
+import java.nio.file.Files
+import java.time.Duration
+
+/**
+ *
+ * @author Paolo Di Tommaso
+ */
+class K8sConfigClientTest extends Specification {
+
+ def 'should create api client with read timeout'() {
+ given:
+ def folder = Files.createTempDirectory('test')
+ def config = folder.resolve('kubeconfig')
+ config.text = '''\
+ apiVersion: v1
+ kind: Config
+ current-context: test
+ clusters:
+ - name: test
+ cluster:
+ server: https://localhost:6443
+ contexts:
+ - name: test
+ context:
+ cluster: test
+ user: test
+ users:
+ - name: test
+ user:
+ token: foo
+ '''.stripIndent()
+ and:
+ def client = new K8sConfigClient(kubeConfigPath: config.toString(), readTimeout: Duration.ofSeconds(45))
+
+ when:
+ def api = client.apiClient()
+ then:
+ api.readTimeout == 45_000
+
+ cleanup:
+ folder?.deleteDir()
+ }
+}
diff --git a/src/test/resources/application-test.yml b/src/test/resources/application-test.yml
index 23b1020d11..b5e1c0b8e7 100644
--- a/src/test/resources/application-test.yml
+++ b/src/test/resources/application-test.yml
@@ -21,6 +21,9 @@ datasources:
schema-generate: CREATE_DROP
---
wave:
+ work-queue:
+ # keep redelivery of RETRY-settled jobs fast so job timeout tests complete within their timeout
+ visibility-timeout: 5s
accounts:
foo: "2cf24dba5fb0a30e26e83b2ac5b9e29e1b161e5c1fa7425e73043362938b9824"
bar: "486ea46224d1bb4fb680f34f7c9ad96a8f24ec88be73ea8e5a6c65260e9cb8a7"