feat: add application-level keepalive to workflow runtime channel

sdk-workflows/pom.xml

@@ -46,6 +46,11 @@
       <groupId>io.opentelemetry</groupId>
       <artifactId>opentelemetry-api</artifactId>
     </dependency>
+    <dependency>
+      <groupId>io.grpc</groupId>
+      <artifactId>grpc-testing</artifactId>
+      <scope>test</scope>
+    </dependency>
   </dependencies>
 
   <build>

sdk-workflows/src/main/java/io/dapr/workflows/internal/GrpcChannelKeepalive.java

@@ -0,0 +1,95 @@
+/*
+ * Copyright 2026 The Dapr Authors
+ * 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.dapr.workflows.internal;
+
+import com.google.protobuf.Empty;
+import io.dapr.durabletask.implementation.protobuf.TaskHubSidecarServiceGrpc;
+import io.grpc.Channel;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.time.Duration;
+import java.util.concurrent.Executors;
+import java.util.concurrent.ScheduledExecutorService;
+import java.util.concurrent.TimeUnit;
+
+/**
+ * Application-level keepalive for a gRPC channel to the Dapr sidecar.
+ *
+ * <p>Periodically invokes the sidecar's {@code hello} RPC so that intermediaries
+ * (e.g. AWS ALBs) that do not treat HTTP/2 PING frames as connection activity
+ * never see the connection as idle and close it.</p>
+ */
+public final class GrpcChannelKeepalive implements AutoCloseable {
+
+  private static final Logger LOGGER = LoggerFactory.getLogger(GrpcChannelKeepalive.class);
+  private static final long KEEPALIVE_DEADLINE_SECONDS = 5;
+
+  private final TaskHubSidecarServiceGrpc.TaskHubSidecarServiceBlockingStub stub;
+  private final String threadName;
+  private final Duration interval;
+  private ScheduledExecutorService scheduler;
+
+  /**
+   * Creates a keepalive for the given channel. No pings are sent until {@link #start()}.
+   *
+   * @param channel    channel to the Dapr sidecar to keep alive.
+   * @param threadName name of the keepalive thread, to tell instances apart in thread dumps.
+   * @param interval   delay between pings.
+   */
+  public GrpcChannelKeepalive(Channel channel, String threadName, Duration interval) {
+    this.stub = TaskHubSidecarServiceGrpc.newBlockingStub(channel);
+    this.threadName = threadName;
+    this.interval = interval;
+  }
+
+  /**
+   * Starts the keepalive loop. Does nothing if already started.
+   */
+  public synchronized void start() {
+    if (this.scheduler != null) {
+      return;
+    }
+    ScheduledExecutorService newScheduler = Executors.newSingleThreadScheduledExecutor(runnable -> {
+      Thread thread = new Thread(runnable, this.threadName);
+      thread.setDaemon(true);
+      return thread;
+    });
+    // The deadline must be applied per ping: Deadline.after snapshots the clock when
+    // withDeadlineAfter is invoked, so hoisting it onto the cached stub would make
+    // every ping after the first fail immediately with DEADLINE_EXCEEDED.
+    // Catch Throwable, not just RuntimeException: any throwable escaping a periodic
+    // task silently cancels all future runs.
+    newScheduler.scheduleWithFixedDelay(() -> {
+      try {
+        this.stub.withDeadlineAfter(KEEPALIVE_DEADLINE_SECONDS, TimeUnit.SECONDS)
+            .hello(Empty.getDefaultInstance());
+      } catch (Throwable e) {
+        LOGGER.debug("Sidecar keepalive ping failed", e);
+      }
+    }, this.interval.toMillis(), this.interval.toMillis(), TimeUnit.MILLISECONDS);
+    this.scheduler = newScheduler;
+  }
+
+  /**
+   * Stops the keepalive loop. Does nothing if not started.
+   */
+  @Override
+  public synchronized void close() {
+    if (this.scheduler != null) {
+      this.scheduler.shutdownNow();
+      this.scheduler = null;
+    }
+  }
+}

sdk-workflows/src/main/java/io/dapr/workflows/runtime/WorkflowRuntime.java

@@ -14,8 +14,11 @@
 package io.dapr.workflows.runtime;
 
 import io.dapr.durabletask.DurableTaskGrpcWorker;
+import io.dapr.workflows.internal.GrpcChannelKeepalive;
 import io.grpc.ManagedChannel;
 
+import javax.annotation.Nullable;
+
 import java.util.concurrent.ExecutorService;
 import java.util.concurrent.TimeUnit;
 
@@ -27,6 +30,7 @@ public class WorkflowRuntime implements AutoCloseable {
   private final DurableTaskGrpcWorker worker;
   private final ManagedChannel managedChannel;
   private final ExecutorService executorService;
+  private final GrpcChannelKeepalive keepalive;
 
   /**
    * Constructor.
@@ -38,9 +42,27 @@ public class WorkflowRuntime implements AutoCloseable {
   public WorkflowRuntime(DurableTaskGrpcWorker worker,
                          ManagedChannel managedChannel,
                          ExecutorService executorService) {
+    this(worker, managedChannel, executorService, null);
+  }
+
+  /**
+   * Constructor.
+   *
+   * @param worker grpcWorker processing activities.
+   * @param managedChannel grpc channel.
+   * @param executorService executor service responsible for running the threads.
+   * @param keepalive application-level keepalive on the worker's channel, started with
+   *                  {@link #start()} and stopped on {@link #close()}. May be null when
+   *                  no keepalive is wanted.
+   */
+  public WorkflowRuntime(DurableTaskGrpcWorker worker,
+                         ManagedChannel managedChannel,
+                         ExecutorService executorService,
+                         @Nullable GrpcChannelKeepalive keepalive) {
     this.worker = worker;
     this.managedChannel = managedChannel;
     this.executorService = executorService;
+    this.keepalive = keepalive;
   }
 
   /**
@@ -57,6 +79,9 @@ public void start() {
    * @param block block the thread if true
    */
   public void start(boolean block) {
+    if (this.keepalive != null) {
+      this.keepalive.start();
+    }
     if (block) {
       this.worker.startAndBlock();
     } else {
@@ -68,6 +93,9 @@ public void start(boolean block) {
    * {@inheritDoc}
    */
   public void close() {
+    if (this.keepalive != null) {
+      this.keepalive.close();
+    }
     this.shutDownWorkerPool();
     this.closeSideCarChannel();
     this.worker.close();

sdk-workflows/src/main/java/io/dapr/workflows/runtime/WorkflowRuntimeBuilder.java

@@ -21,6 +21,7 @@
 import io.dapr.workflows.Workflow;
 import io.dapr.workflows.WorkflowActivity;
 import io.dapr.workflows.internal.ApiTokenClientInterceptor;
+import io.dapr.workflows.internal.GrpcChannelKeepalive;
 import io.grpc.ClientInterceptor;
 import io.grpc.ManagedChannel;
 import org.apache.commons.lang3.StringUtils;
@@ -43,6 +44,7 @@ public class WorkflowRuntimeBuilder {
   private final Set<String> workflowSet = Collections.synchronizedSet(new HashSet<>());
   private final DurableTaskGrpcWorkerBuilder builder;
   private final ManagedChannel managedChannel;
+  private final Properties properties;
   private ExecutorService executorService;
 
   /**
@@ -69,6 +71,7 @@ private WorkflowRuntimeBuilder(Properties properties, Logger logger) {
     this.workflowApiTokenInterceptor = new ApiTokenClientInterceptor(properties);
     this.managedChannel = NetworkUtils.buildGrpcManagedChannel(properties, workflowApiTokenInterceptor);
     this.builder = new DurableTaskGrpcWorkerBuilder().grpcChannel(this.managedChannel);
+    this.properties = properties;
     this.logger = logger;
   }
 
@@ -82,9 +85,14 @@ public WorkflowRuntime build() {
       synchronized (WorkflowRuntime.class) {
         this.executorService = this.executorService == null ? Executors.newCachedThreadPool() : this.executorService;
         if (instance == null) {
+          GrpcChannelKeepalive keepalive = null;
+          if (this.properties.getValue(Properties.WORKFLOWS_RUNTIME_APP_KEEP_ALIVE_ENABLED)) {
+            keepalive = new GrpcChannelKeepalive(this.managedChannel, "dapr-workflow-runtime-keepalive",
+                this.properties.getValue(Properties.WORKFLOWS_APP_KEEP_ALIVE_INTERVAL_SECONDS));
+          }
           instance = new WorkflowRuntime(
               this.builder.withExecutorService(this.executorService).build(),
-              this.managedChannel, this.executorService);
+              this.managedChannel, this.executorService, keepalive);
         }
       }
     }

sdk-workflows/src/test/java/io/dapr/workflows/internal/GrpcChannelKeepaliveTest.java

@@ -0,0 +1,148 @@
+/*
+ * Copyright 2026 The Dapr Authors
+ * 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.dapr.workflows.internal;
+
+import com.google.protobuf.Empty;
+import io.dapr.durabletask.implementation.protobuf.TaskHubSidecarServiceGrpc;
+import io.grpc.ManagedChannel;
+import io.grpc.Server;
+import io.grpc.Status;
+import io.grpc.inprocess.InProcessChannelBuilder;
+import io.grpc.inprocess.InProcessServerBuilder;
+import io.grpc.stub.StreamObserver;
+import org.junit.jupiter.api.AfterEach;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+
+import java.io.IOException;
+import java.time.Duration;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.concurrent.atomic.AtomicInteger;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+public class GrpcChannelKeepaliveTest {
+
+  private static final Duration TEST_INTERVAL = Duration.ofMillis(50);
+
+  private final AtomicInteger helloCount = new AtomicInteger();
+  private final AtomicBoolean failPings = new AtomicBoolean(false);
+
+  private Server server;
+  private ManagedChannel channel;
+
+  @BeforeEach
+  public void setUp() throws IOException {
+    String serverName = InProcessServerBuilder.generateName();
+    this.server = InProcessServerBuilder.forName(serverName)
+        .directExecutor()
+        .addService(new TaskHubSidecarServiceGrpc.TaskHubSidecarServiceImplBase() {
+          @Override
+          public void hello(Empty request, StreamObserver<Empty> responseObserver) {
+            helloCount.incrementAndGet();
+            if (failPings.get()) {
+              responseObserver.onError(Status.UNAVAILABLE.asRuntimeException());
+              return;
+            }
+            responseObserver.onNext(Empty.getDefaultInstance());
+            responseObserver.onCompleted();
+          }
+        })
+        .build()
+        .start();
+    this.channel = InProcessChannelBuilder.forName(serverName).directExecutor().build();
+  }
+
+  @AfterEach
+  public void tearDown() {
+    this.channel.shutdownNow();
+    this.server.shutdownNow();
+  }
+
+  @Test
+  public void noPingsBeforeStart() throws InterruptedException {
+    try (GrpcChannelKeepalive keepalive = new GrpcChannelKeepalive(channel, "test-keepalive", TEST_INTERVAL)) {
+      Thread.sleep(TEST_INTERVAL.toMillis() * 4);
+      assertEquals(0, helloCount.get());
+    }
+  }
+
+  @Test
+  public void pingsPeriodically() throws InterruptedException {
+    try (GrpcChannelKeepalive keepalive = new GrpcChannelKeepalive(channel, "test-keepalive", TEST_INTERVAL)) {
+      keepalive.start();
+      // A second start must not schedule a second ping loop.
+      keepalive.start();
+      awaitHelloCountAtLeast(2);
+    }
+  }
+
+  @Test
+  public void continuesPingingAfterFailures() throws InterruptedException {
+    failPings.set(true);
+    try (GrpcChannelKeepalive keepalive = new GrpcChannelKeepalive(channel, "test-keepalive", TEST_INTERVAL)) {
+      keepalive.start();
+      awaitHelloCountAtLeast(2);
+      failPings.set(false);
+      awaitHelloCountAtLeast(helloCount.get() + 2);
+    }
+  }
+
+  @Test
+  public void closeStopsPinging() throws InterruptedException {
+    GrpcChannelKeepalive keepalive = new GrpcChannelKeepalive(channel, "test-keepalive", TEST_INTERVAL);
+    keepalive.start();
+    awaitHelloCountAtLeast(1);
+    keepalive.close();
+    // Allow an already in-flight ping to finish before snapshotting.
+    Thread.sleep(TEST_INTERVAL.toMillis() * 2);
+    int countAfterClose = helloCount.get();
+    Thread.sleep(TEST_INTERVAL.toMillis() * 4);
+    assertEquals(countAfterClose, helloCount.get());
+  }
+
+  @Test
+  public void keepaliveThreadIsDaemonAndStopsOnClose() throws InterruptedException {
+    String threadName = "test-keepalive-lifecycle";
+    try (GrpcChannelKeepalive keepalive = new GrpcChannelKeepalive(channel, threadName, TEST_INTERVAL)) {
+      keepalive.start();
+      awaitHelloCountAtLeast(1);
+      Thread thread = findThread(threadName);
+      assertTrue(thread != null && thread.isDaemon(), "expected a live daemon keepalive thread");
+    }
+    long deadline = System.currentTimeMillis() + TimeUnit.SECONDS.toMillis(5);
+    while (findThread(threadName) != null && System.currentTimeMillis() < deadline) {
+      Thread.sleep(10);
+    }
+    assertTrue(findThread(threadName) == null, "keepalive thread should terminate after close()");
+  }
+
+  private void awaitHelloCountAtLeast(int expected) throws InterruptedException {
+    long deadline = System.currentTimeMillis() + TimeUnit.SECONDS.toMillis(5);
+    while (helloCount.get() < expected && System.currentTimeMillis() < deadline) {
+      Thread.sleep(10);
+    }
+    assertTrue(helloCount.get() >= expected,
+        "expected at least " + expected + " hello pings, got " + helloCount.get());
+  }
+
+  private static Thread findThread(String name) {
+    return Thread.getAllStackTraces().keySet().stream()
+        .filter(t -> t.getName().equals(name) && t.isAlive())
+        .findFirst()
+        .orElse(null);
+  }
+}